@lishenxydlgzs/aws-athena-mcp

一个用于运行 AWS Athena 查询的 Model Context Protocol (MCP) 服务器。该服务器使 AI 助手能够对您的 AWS Athena 数据库执行 SQL 查询并检索结果。

使用方法

1. 使用以下任一方法配置 AWS 凭证：

   • AWS CLI 配置
   • 环境变量 (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
   • IAM 角色（如果在 AWS 上运行）

2. 将服务器添加到您的 MCP 配置中：

{
  "mcpServers": {
    "athena": {
      "command": "npx",
      "args": ["-y", "@lishenxydlgzs/aws-athena-mcp"],
      "env": {
        // Required
        "OUTPUT_S3_PATH": "s3://your-bucket/athena-results/",
        
        // Optional AWS configuration
        "AWS_REGION": "us-east-1",                    // Default: AWS CLI default region
        "AWS_PROFILE": "default",                     // Default: 'default' profile
        "AWS_ACCESS_KEY_ID": "",                      // Optional: AWS access key
        "AWS_SECRET_ACCESS_KEY": "",                  // Optional: AWS secret key
        "AWS_SESSION_TOKEN": "",                      // Optional: AWS session token
        
        // Optional server configuration
        "ATHENA_WORKGROUP": "default_workgroup",      // Optional: specify the Athena WorkGroup
        "QUERY_TIMEOUT_MS": "300000",                 // Default: 5 minutes (300000ms)
        "MAX_RETRIES": "100",                         // Default: 100 attempts
        "RETRY_DELAY_MS": "500"                       // Default: 500ms between retries
      }
    }
  }
}

1. 服务器提供以下工具：

• run_query: 使用 AWS Athena 执行 SQL 查询

  • 参数：
    • database: 要查询的 Athena 数据库
    • query: 要执行的 SQL 查询
    • maxRows: 返回的最大行数（默认：1000，最大：10000）
  • 返回：
    • 如果查询在超时前完成：完整的查询结果
    • 如果达到超时：仅返回 queryExecutionId 以供稍后检索

• get_status: 检查查询执行的状态

  • 参数：
    • queryExecutionId: 从 run_query 返回的 ID
  • 返回：
    • state: 查询状态（QUEUED, RUNNING, SUCCEEDED, FAILED, 或 CANCELLED）
    • stateChangeReason: 状态更改的原因（如果有）
    • submissionDateTime: 查询提交的时间
    • completionDateTime: 查询完成的时间（如果已完成）
    • statistics: 查询执行统计信息（如果可用）

• get_result: 检索已完成查询的结果

  • 参数：
    • queryExecutionId: 从 run_query 返回的 ID
    • maxRows: 返回的最大行数（默认：1000，最大：10000）
  • 返回：
    • 如果查询成功完成，则返回完整查询结果
    • 如果查询失败或仍在运行，则返回错误

• list_saved_queries: 列出 Athena 中所有保存的（命名）查询。

• 返回：

  • 一个包含 id、name 和可选 description 的保存查询数组
  • 查询从配置的 ATHENA_WORKGROUP 和 AWS_REGION 中返回

• run_saved_query: 通过其 ID 运行之前保存的查询。

• 参数：

  • namedQueryId: 保存查询的 ID
  • databaseOverride: 可选地覆盖保存查询的默认数据库
  • maxRows: 返回的最大行数（默认：1000）
  • timeoutMs: 超时时间（毫秒，默认：60000）

• 返回：

  • 与 run_query 相同的行为：完整结果或执行 ID

使用示例

显示所有数据库

发送给 AI 助手的消息：
列出 Athena 中的所有数据库

MCP 参数：

{
  "database": "default",
  "query": "SHOW DATABASES"
}

列出数据库中的表

发送给 AI 助手的消息：
显示默认数据库中的所有表

MCP 参数：

{
  "database": "default",
  "query": "SHOW TABLES"
}

获取表结构

发送给 AI 助手的消息：
asin_sitebestimg 表的结构是什么？

MCP 参数：

{
  "database": "default",
  "query": "DESCRIBE default.asin_sitebestimg"
}

表行预览

向 AI 助手发送的消息：
Show some rows from my_database.mytable

MCP 参数：

{
  "database": "my_database",
  "query": "SELECT * FROM my_table LIMIT 10",
  "maxRows": 10
}

带有过滤和聚合的高级查询

向 AI 助手发送的消息：
Find the average price by category for in-stock products

MCP 参数：

{
  "database": "my_database",
  "query": "SELECT category, COUNT(*) as count, AVG(price) as avg_price FROM products WHERE in_stock = true GROUP BY category ORDER BY count DESC",
  "maxRows": 100
}

检查查询状态

{
  "queryExecutionId": "12345-67890-abcdef"
}

获取已完成查询的结果

{
  "queryExecutionId": "12345-67890-abcdef",
  "maxRows": 10
}

列出保存的查询

{
  "name": "list_saved_queries",
  "arguments": {}
}

运行保存的查询

{
  "name": "run_saved_query",
  "arguments": {
    "namedQueryId": "abcd-1234-efgh-5678",
    "maxRows": 100
  }
}

要求

• Node.js >= 16
• 具有适当 Athena 和 S3 权限的 AWS 凭证
• 用于存储查询结果的 S3 存储桶
• 命名查询（可选）必须存在于指定的 ATHENA_WORKGROUP 和 AWS_REGION 中

许可证

MIT

仓库

GitHub 仓库