1. 为什么要在本地把 MCP Servers 接到 Amazon S3 元数据表
Amazon S3 元数据表是什么？简单说，它把 S3 桶里每个对象的创建、更新、删除事件，自动写进一张完全托管的 Apache Iceberg 表。你不用去遍历对象，直接写 SQL 就能查“谁在什么时候传了什么、用了哪种存储类、有没有加密”。它适合谁？适合那些桶里已经堆了几十万甚至上百万对象、想快速做数据洞察又不想自己维护元数据管道的开发者。
我试过最直接的痛点：以前想统计某个桶里 REDUCED_REDUNDANCY 存储类的对象有多少，得先 list 对象、再逐个 head_object，脚本跑十几分钟还容易限流。有了元数据表，一条GROUP BY storage_class的 SQL 几秒出结果。但问题来了——每次都要打开 Athena 控制台、手写 SQL、复制结果，来回切换很打断思路。
MCP Servers 正好补上这一段。MCP（Model Context Protocol）可以理解成 AI 应用的 USB-C 接口：它把“查询 S3 元数据表”这件事封装成标准工具，让 Claude Code 这类客户端用自然语言就能调用。你问“按 bucket 分组统计对象数”，它自动生成 Athena SQL、执行、把结果整理回来。整条链路是：本地 MCP Client → MCP Server（stdio）→ boto3 → Amazon Athena → S3 元数据表（Iceberg）。
这篇要交付的就是一条能跑通的本地链路：可复制的 MCP Server 配置片段、S3 元数据表查询示例、以及端到端验证动作。下面从环境准备开始，一步步来。
2. 前置准备：Athena 查询 S3 元数据表与 MCP Server 环境搭建
在写 MCP 配置之前，先把底层跑通。S3 元数据表本身要通过 Athena 查询，所以顺序是：先确认元数据表已配置、Athena 能查到，再把它包成 MCP Server。
第一步，确认元数据表状态。在 S3 控制台找到目标桶，进入“元数据”配置，确认元数据表已创建。表名通常形如s3metadata_<bucket>_<region>，比如s3metadata_imagebucket_us_east_1。记下它的 catalog、database、table 三个值，后面配置要用。
第二步，在 Athena 里验证能查。打开 Athena 查询编辑器，切到对应 catalog（一般是s3tablescatalog），执行一条最简单的查询：
SELECT bucket, key, record_type, record_timestamp, size, storage_class FROM aws_s3_metadata.s3metadata_imagebucket_us_east_1 LIMIT 10;能返回行，说明元数据表、Athena 集成、输出位置都正常。如果这里就报错，先别往下走，去检查元数据表配置和 Athena 结果输出桶。
第三步，准备本地 Python 环境。MCP Server 用 Python 写，依赖mcp、boto3、httpx。建议用 uv 管理，干净利落：
uv init aws_s3table_query cd aws_s3table_query uv add mcp boto3 httpx第四步，配置 AWS 凭证。MCP Server 通过 boto3 调 Athena，需要本地有可用的凭证。用 AWS CLI 配置：
aws configure # 依次输入 Access Key、Secret Key、region（如 us-east-1）、输出格式 json注意，这个 IAM 用户或角色需要两个权限：一是 Athena 的查询权限（athena:StartQueryExecution、athena:GetQueryExecution、athena:GetQueryResults），二是对结果输出桶的读写权限。如果还要用 Bedrock 上的 Claude 模型，再补 Bedrock 调用权限。
第五步，把 Server 代码放进目录。核心结构就是前面 excerpt 里的那套：FastMCP("aws_s3table_query")初始化、execute_query执行 Athena、query_record和query_statistics两个@mcp.tool()。文件保存为server.py，放在aws_s3table_query目录下。
到这里，底层链路和 Server 代码都就位了。接下来是关键的配置环节——把 Server 注册到 MCP Client。
3. 可复制配置：Claude Code 接入 S3 元数据表 MCP Server 的完整 settings 片段
这一节给可直接复制的配置。以 Claude Code 作为 MCP Client，它支持用claude mcp add-json注册 stdio 类型的 Server。
先装 Claude Code：
npm install -g @anthropic-ai/claude-code如果你走 Bedrock 上的 Claude 模型，设置两个环境变量：
export CLAUDE_CODE_USE_BEDROCK=1 export ANTHROPIC_MODEL='us.anthropic.claude-3-7-sonnet-20250219-v1:0'然后注册 MCP Server。下面这段 JSON 是完整配置，路径和环境变量按你的实际情况替换：
{ "type": "stdio", "command": "uv", "args": [ "--directory", "/Users/xxx/mcp-server-python/aws_s3table_query", "run", "server.py" ], "env": { "ATHENA_CATALOG": "s3tablescatalog", "ATHENA_DATABASE": "aws_s3_metadata", "ATHENA_TABLE": "s3metadata_imagebucket_us_east_1", "ATHENA_OUTPUT_LOCATION": "s3://aws-athena-xxx-us-east-1/", "AWS_REGION": "us-east-1" } }用一条命令注册：
claude mcp add-json s3table-mcp-server '{"type":"stdio","command":"uv","args":["--directory","/Users/xxx/mcp-server-python/aws_s3table_query","run","server.py"],"env":{"ATHENA_CATALOG":"s3tablescatalog","ATHENA_DATABASE":"aws_s3_metadata","ATHENA_TABLE":"s3metadata_imagebucket_us_east_1","ATHENA_OUTPUT_LOCATION":"s3://aws-athena-xxx-us-east-1/","AWS_REGION":"us-east-1"}}'需要替换的部分，逐个说清楚：
s3table-mcp-server是 Server 配置名，可自定义，后面claude mcp list会显示这个名字。
/Users/xxx/mcp-server-python/aws_s3table_query是你本地 Server 代码目录的绝对路径，server.py就在这个目录下。
ATHENA_CATALOG填 Athena 目录名，S3 元数据表场景一般是s3tablescatalog。
ATHENA_DATABASE填数据库名，通常是aws_s3_metadata。
ATHENA_TABLE填你要查的元数据表名，注意区分桶和 region。
ATHENA_OUTPUT_LOCATION是 Athena 查询结果的 S3 输出位置，必须以斜杠结尾，否则 Athena 会报输出路径无效。
AWS_REGION建议显式写上，避免 boto3 取到意外区域。
这里有个容易忽略的点：uv run server.py会以 stdio 方式启动 Server，Claude Code 通过标准输入输出和它通信。所以server.py里mcp.run(transport='stdio')这行不能改。如果你之前写的是 SSE 或 HTTP transport，Claude Code 这边会连不上。
配置写完后，用claude mcp list验证。看到s3table-mcp-server出现在列表里、状态正常，就说明注册成功。如果显示连接失败，先手动跑一次uv run server.py，看有没有 Python 报错——环境变量缺失、boto3 凭证问题都会在这一步暴露。
4. 验证请求：用自然语言查询 S3 元数据表并拿到成功结果
配置就位后，进入验证环节。启动 Claude Code：
claude在交互界面里，直接用自然语言提问。第一个用例，查记录类型为 CREATE 的对象：
查询 s3table 记录类型是 CREATE 的记录Claude Code 会调用query_record工具，record_type参数传CREATE。Server 内部拼出这样的 SQL：
SELECT bucket, key, sequence_number, record_type, record_timestamp, size, last_modified_date, e_tag, storage_class, is_multipart, encryption_status, is_bucket_key_enabled, kms_key_arn, checksum_algorithm, object_tags, user_metadata, requester, source_ip_address, request_id FROM aws_s3_metadata.s3metadata_imagebucket_us_east_1 WHERE record_type = 'CREATE'Athena 执行后，结果以列表形式返回。你会看到每个对象的 bucket、key、大小、存储类、加密状态等字段。这一步成功，说明从 MCP Client 到 Athena 的整条链路通了。
第二个用例，按 bucket 分组统计：
按照 bucket 分组帮我统计这次调用query_statistics，group_by传["bucket"]。生成的 SQL 是：
SELECT bucket, COUNT(*) as total_objects, SUM(size) as total_size, COUNT(DISTINCT record_type) as unique_record_types, COUNT(DISTINCT storage_class) as unique_storage_classes FROM aws_s3_metadata.s3metadata_imagebucket_us_east_1 GROUP BY bucket返回结果里，每个 bucket 对应一行，包含对象总数、总大小、记录类型种类数、存储类种类数。这个结果直接回答了“哪个桶占了多少空间”这类数据洞察问题。
再试一个带过滤条件的组合查询，比如查某个时间范围、某个存储类的对象：
查询 2025-01-01 之后、存储类是 STANDARD 的记录Server 的build_where_clause会把start_time和storage_class拼进 WHERE 子句，Athena 返回匹配行。整个过程你不需要手写一行 SQL。
验证成功的标志有三个：一是 Claude Code 明确显示调用了query_record或query_statistics工具；二是返回结果里有真实数据行，不是空列表；三是没有抛异常。如果返回空列表，可能是过滤条件太严，或者元数据表里确实没有对应记录，可以放宽条件再试。
5. 常见报错排查：401、local proxy failed、reading choices、OAuth 逐条对照
链路跑通前，大概率会撞上几个典型报错。这一节按真实错误信息逐条对照。
401 Unauthorized / AccessDeniedException。这个通常出在 Athena 或 S3 权限上。MCP Server 用本地 AWS 凭证调 Athena，如果 IAM 用户没有athena:StartQueryExecution，就会 401。排查方法：在终端直接跑aws athena start-query-execution --query-string "SELECT 1" --result-configuration OutputLocation=s3://your-bucket/，看是否报权限错误。如果是，去 IAM 补策略。另外确认ATHENA_OUTPUT_LOCATION指向的桶，当前凭证有写权限。
local proxy failed / connection refused。这个报错一般和 MCP Client 与 Server 的通信方式有关。Claude Code 用 stdio 和 Server 通信，如果你在server.py里写成了mcp.run(transport='sse')，Client 按 stdio 连就会失败。检查server.py最后一行，确保是mcp.run(transport='stdio')。还有一种情况是uv --directory路径写错，Server 根本没启动，Client 连不上。手动执行uv --directory /your/path run server.py，看能否正常启动。
Error reading choices / 返回结构解析失败。这个多出现在模型侧返回格式异常时。如果你走 Bedrock 的 Claude 模型，确认ANTHROPIC_MODEL的 model id 拼写正确，比如us.anthropic.claude-3-7-sonnet-20250219-v1:0。model id 错了，Bedrock 返回的错误结构不是标准 choices 格式，Client 解析就报 reading choices。另外确认该 region 支持你选的模型。
OAuth / 凭证过期。如果你用的是临时凭证（STS），过期后会报 OAuth 相关或ExpiredToken。重新执行aws configure或刷新临时凭证即可。长期跑的话，建议用有长期凭证的 IAM 用户，或者配置凭证刷新机制。
Athena 查询超时或 FAILED。wait_for_query_completion会轮询查询状态，如果 Athena 侧查询失败，会抛Query xxx failed with state FAILED。常见原因是 SQL 语法错误、表名不存在、catalog 写错。把 Server 里print(query)打出的 SQL 复制到 Athena 控制台手动执行，能快速定位。
环境变量缺失。Server 启动时会校验ATHENA_DATABASE、ATHENA_TABLE、ATHENA_OUTPUT_LOCATION，缺任何一个都会抛Missing required environment variables。检查claude mcp add-json里的env字段是否完整，注意 JSON 转义。
排查顺序建议：先手动跑 Server 看 Python 层报错，再在终端验证 AWS 凭证和 Athena 权限，最后检查 MCP Client 配置。三层分开定位，比一上来就猜快得多。
6. 把 S3 元数据洞察接进日常：从查询到 Coding Plan 的落地路径
链路跑通后，真正的价值在于把它变成日常动作。几个我实际用下来的场景：内容团队想知道某个图片桶里哪些对象缺user_metadata['location']，直接问 Claude Code，它调query_record加自定义过滤；安全团队要审计未加密对象，问一句“列出 encryption_status 为空的记录”，结果直接出来；成本优化时按 storage_class 分组统计，一眼看出哪些桶该转 INTELLIGENT_TIERING。
如果你想让这套能力更稳定地服务长期编码和 Agent 任务，可以考虑 TaoToken 的 Coding Plan，把模型调用和工具链统一管理起来。配置入口在 https://taotoken.net/api，API Keys 在 https://taotoken.net/api-keys 管理，接入文档在 https://taotoken.net/doc。模型对话可以直接在 https://taotoken.net/chat 验证，控制台在 https://taotoken.net/console。Claude Code 相关的接入说明在 https://taotoken.net/claude-code。
回到技术本身，还有两个可以继续做的方向。一是扩展工具，比如加一个find_objects_by_metadata，按用户自定义元数据条件发现对象；二是加合规检查工具，自动扫描未加密或存储类不当的对象。这些都可以在现有server.py基础上加@mcp.tool()实现，不用改架构。
最后提醒一句：元数据表里包含对象标签、请求者身份、源 IP 这类敏感信息，查询权限要收紧，别让不该看的人拿到。表维护和查询成本也要留意，Athena 按扫描量计费，查询时尽量加过滤条件、只选需要的列。
