AI导哈

Athena MCP服务器

一个用于管理Amazon Athena资源的模型上下文协议（MCP）服务器，支持执行SQL查询、管理命名查询、工作组和数据目录。

  • by Moonlight-CL
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 moonlight-cl/athena-mcp-server

使用说明

Athena MCP服务器

AWS Athena的模型上下文协议（MCP）服务器，提供管理Amazon Athena资源的工具。

这个项目的灵感来自 aws数据处理mcp服务器 并调整其代码以创建支持流式HTTP传输类型的专用Athena MCP服务器。

特性

  • 执行和管理Athena SQL查询
  • 创建和管理命名查询
  • 管理Athena工作组
  • 与Athena数据目录交互
  • 支持stdio和Streamable HTTP传输类型

项目结构

athena-mcp-server/
├── main.py                           # Main entry point
├── pyproject.toml                    # Project configuration
├── README.md                         # Documentation
└── athena_mcp_server/
    ├── server.py                     # MCP server implementation
    ├── handlers/
    │   ├── __init__.py
    │   ├── athena_query_handler.py        # Query execution and named queries
    │   ├── athena_data_catalog_handler.py # Data catalog operations
    │   └── athena_workgroup_handler.py    # Workgroup operations
    ├── models/
    │   ├── __init__.py
    │   └── athena_models.py          # Response models
    └── utils/
        ├── __init__.py
        ├── aws_helper.py             # AWS client utilities
        └── logging_helper.py         # Logging utilities

安装

# Clone the repository
git clone https://github.com/yourusername/athena-mcp-server.git
cd athena-mcp-server

# Install dependencies using uv
uv sync

用法

启动服务器

# Start the server in read-only mode (default)
uv run main.py

# Start the server with write access
uv run main.py --allow-write

# Start the server with HTTP transport (default is stdio)
uv run main.py --transport http

# Start the server with custom host and port
uv run main.py --host 127.0.0.1 --port 8080

环境变量

  • AWS_REGION：用于AWS API调用的AWS区域
  • AWS_PROFILE：用于凭据的AWS配置文件
  • FASTMCP_LOG_LEVEL：日志级别（默认值：警告）

可用工具

查询执行

  • manage_aws_athena_query_executions：执行和管理Athena SQL查询
    • 操作：批处理查询执行、查询执行、获取查询结果、获取查询运行时统计、列出查询执行、启动查询执行、停止查询执行

命名查询

  • manage_aws_athena_named_queries：管理Athena中保存的SQL查询
    • 操作：批处理命名查询、创建命名查询、删除命名查询、获取命名查询、列出命名查询、更新命名查询

数据目录

  • manage_aws_athena_data_catalogs：管理Athena数据目录
    • 操作：创建数据目录、删除数据目录、获取数据目录、列出数据目录、更新数据目录

工作组

  • manage_aws_athena_workgroups：管理Athena工作组
    • 操作：创建工作组、删除工作组、获取工作组、列出工作组、更新工作组

例子

运行查询

# Start a new query
response = await manage_aws_athena_query_executions(
    operation='start-query-execution',
    query_string='SELECT * FROM my_database.my_table LIMIT 10',
    query_execution_context={'Database': 'my_database', 'Catalog': 'my_catalog'},
    work_group='primary',
)

# Get the query results
results = await manage_aws_athena_query_executions(
    operation='get-query-results', query_execution_id=response.query_execution_id
)

创建命名查询

# Create a named query
create_response = await manage_aws_athena_named_queries(
    operation='create-named-query',
    name='Daily Active Users',
    description='Query to calculate daily active users',
    database='analytics',
    query_string='SELECT date, COUNT(DISTINCT user_id) AS active_users FROM user_events GROUP BY date ORDER BY date DESC',
    work_group='primary',
)

# Later, retrieve the named query
query = await manage_aws_athena_named_queries(
    operation='get-named-query', named_query_id=create_response.named_query_id
)

管理工作组

# Create a workgroup with cost controls
create_response = await manage_aws_athena_workgroups(
    operation='create-work-group',
    work_group_name='data-science-team',
    description='Workgroup for data science team',
    configuration={
        'ResultConfiguration': {'OutputLocation': 's3://my-bucket/athena-results/'},
        'EnforceWorkGroupConfiguration': True,
        'BytesScannedCutoffPerQuery': 10737418240,  # 10GB
        'PublishCloudWatchMetricsEnabled': True,
    },
)

# List all workgroups
workgroups = await manage_aws_athena_workgroups(operation='list-work-groups')

许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

相关 MCP