基于AWS Bedrock AgentCore的MCP服务器实战：构建AI工具集成标准接口
·
1. 项目概述：为什么MCP是AI架构的“USB-C”接口？
如果你最近在折腾AI应用，尤其是想让Claude、Cursor这类智能体（Agent）去操作你的AWS云资源，比如查一下DynamoDB里的用户数据，或者看看S3桶里存了多少文件，那你大概率已经遇到了一个核心难题： 工具集成 。传统做法是什么？要么你把调用AWS SDK的代码硬编码到提示词（Prompt）里，让模型去“猜”着调用，结果往往是模型“幻觉”出根本不存在的函数；要么你写一堆复杂的中间件和API网关，把工具调用逻辑封装起来，但每对接一个新模型或新场景，就得重写一遍适配层，既脆弱又难以维护。
这就是为什么我说， MCP（Model Context Protocol） 是2026年最重要的AI基础设施模式，没有之一。它由Anthropic提出，现在由Linux基金会管理，而AWS更是将其提升为Bedrock AgentCore中的一等公民。你可以把它理解为AI工具领域的 USB-C标准 。以前，每个设备（AI模型）都有自己的充电口（工具调用方式），互不兼容。现在，MCP定义了一个通用协议，让你开发的工具服务器（MCP Server）可以像USB-C设备一样，即插即用在任何支持MCP的AI客户端（Claude Desktop、Cursor、Bedrock Agent）上。
想象一下这个场景：你为团队内部开发了一个能查询生产数据库的工具。没有MCP之前，你需要为Claude写一套提示词，为Cursor再写一套，如果以后换用其他模型，还得重来。有了MCP，你只需要用Python（或其他语言）写好一个MCP服务器，定义好工具的函数和描述。任何接入MCP的AI智能体都能自动发现这些工具，理解它们的用途，并在合适的时机调用它们。这彻底将 工具的实现 与 工具的使用 解耦了。
更关键的是，AWS Bedrock AgentCore Runtime的推出，让MCP服务器从需要你自己运维的容器服务，变成了一个 全托管的、无服务器（Serverless）的云服务 。这意味着你不再需要操心服务器的扩缩容、会话隔离、网络配置，AWS帮你全包了。本文的目标，就是跳过所有理论，用大约30分钟，手把手带你部署一个能连接真实AWS服务（DynamoDB和S3）的MCP服务器到Bedrock上，并让一个Claude智能体真正用起来。这不是概念验证，而是包含完整代码、部署脚本和避坑指南的生产级实践。
2. 核心思路与架构设计：从本地测试到云端托管
在动手写代码之前，我们先拆解一下整个系统的架构，理解每个环节的设计考量。这能帮助你在未来扩展或排查问题时，清楚地知道数据流向了哪里。
2.1 MCP的核心工作流：一次完整的工具调用是如何发生的？
整个流程围绕着“定义工具、发现工具、使用工具”展开，涉及三个核心角色：
-
MCP服务器（Server） ：这是我们即将构建的核心。它对外暴露一组定义好的工具（比如
query_dynamodb）。它的职责是：- 向客户端宣告自己有哪些工具可用（提供工具的名称、描述、参数JSON Schema）。
- 接收来自客户端的工具调用请求，执行相应的业务逻辑（如查询数据库）。
- 将执行结果（成功或失败）格式化成标准响应，返回给客户端。
-
MCP客户端（Client） ：通常内置于AI应用内部，如Claude Desktop、Cursor或Bedrock Agent。它的职责是：
- 连接到一个或多个MCP服务器。
- 从服务器获取工具列表及其描述。
- 在AI模型进行推理时，根据用户的问题和上下文，判断是否需要、以及需要调用哪个工具。
- 将模型生成的工具调用指令转发给对应的MCP服务器，并等待结果。
- 将工具返回的结果整合进对话上下文，供模型生成最终回答。
-
AI智能体（Agent） ：通常指大型语言模型（如Claude）及其提示词工程构成的系统。它利用MCP客户端提供的工具能力来完成任务。
一次典型的交互流程如下：
用户提问: “我的‘data-lake-prod’ S3桶里有多少个文件？”
↓
Bedrock Agent (Claude模型) 分析问题，通过MCP客户端发现可用的 `get_s3_summary` 工具。
↓
Claude决定调用该工具，并生成符合工具Schema的调用参数：`{“bucket_name”: “data-lake-prod”}`。
↓
MCP客户端将调用请求发送给我们的MCP服务器。
↓
MCP服务器执行 `get_s3_summary` 函数，使用boto3 SDK向AWS S3发起API调用。
↓
AWS S3返回文件列表。
↓
MCP服务器将结果格式化为JSON字符串，返回给客户端。
↓
MCP客户端将结果（“桶内有150个文件，总计2.3GB”）插入对话历史。
↓
Claude模型基于这个新信息，生成最终回答：“您的‘data-lake-prod’桶内共有150个文件，总大小约为2.3GB。”
这个架构的美妙之处在于 关注点分离 。作为工具开发者，你只需要关心如何用Python和boto3安全、高效地操作AWS服务。至于AI模型如何理解、何时调用你的工具，那是MCP客户端和模型自身的事情。这种标准化极大地提升了开发效率和系统的可维护性。
2.2 方案选型：为什么选择FastMCP和Bedrock AgentCore？
在构建MCP服务器时，你有几个选择：直接使用原始的MCP SDK处理底层的JSON-RPC协议，或者使用更高级的封装框架。我强烈推荐使用 FastMCP 。
- 原始MCP SDK ：你需要手动处理初始化、工具注册、请求解析、响应封装等大量样板代码。虽然能让你更深入理解协议，但对于快速开发和维护来说，性价比极低。
- FastMCP ：它是一个Python框架，用装饰器（如
@mcp.tool()）抽象了所有协议细节。你只需要像写普通Python函数一样定义工具，FastMCP会自动从你的函数签名和文档字符串生成MCP所需的完整JSON Schema。这让你能专注于业务逻辑，而不是协议通信。
在部署层面，传统方式可能是将MCP服务器代码部署在EC2实例或ECS/Fargate服务上，然后自己管理负载均衡、会话和生命周期。而现在， AWS Bedrock AgentCore Runtime 提供了更优解。
- 传统自托管 ：你需要自己配置Docker镜像、设置VPC、管理安全组、处理扩缩容策略，并为可能出现的会话冲突（多个用户共享服务器状态）编写复杂的逻辑。
- Bedrock AgentCore Runtime ：这是一个专为运行MCP服务器设计的全托管环境。你只需要提供一个Docker镜像。AgentCore会为 每个用户会话自动创建一个独立的微虚拟机（microVM） ，实现了天然的会话隔离。它自动处理伸缩、监控、日志集成，并提供了长达8小时的会话支持，非常适合需要保持状态的长时运行任务。选择它，意味着你将运维复杂度降到了最低。
2.3 环境与权限准备：安全第一
在开始编码前，确保你的本地开发环境已就绪。你需要：
- Python 3.11+ ：这是FastMCP的推荐版本。可以使用
pyenv或conda管理多版本Python。 - AWS CLI 已配置 ：在终端运行
aws configure，确保已设置好具有编程访问权限的IAM用户的Access Key和Secret Key，并选择好区域（如us-east-1）。 - 必要的IAM权限 ：用于本地测试和部署的IAM用户或角色，需要至少具备以下权限：
- 对于S3 ：
s3:ListBucket,s3:GetObject（针对你计划查询的桶）。 - 对于DynamoDB ：
dynamodb:GetItem,dynamodb:Query,dynamodb:Scan（根据你的查询需求）。 - 对于ECR ：
ecr:GetAuthorizationToken,ecr:CreateRepository,ecr:BatchGetImage,ecr:PutImage（用于推送Docker镜像）。 - 对于Bedrock AgentCore ：
bedrock-agentcore:CreateAgentRuntime等（通常通过托管策略AmazonBedrockFullAccess或更细粒度的自定义策略授予）。
重要安全提示 ：遵循最小权限原则。在生产环境中，切勿使用拥有
AdministratorAccess的凭证进行开发。为这个MCP服务器专门创建一个IAM角色，并只附加其必需的具体权限。 - 对于S3 ：
3. 手把手构建MCP服务器：从代码到本地测试
理论讲完，我们进入实战环节。我们将构建一个提供两个核心工具的MCP服务器：查询DynamoDB和总结S3桶。
3.1 初始化项目与安装依赖
首先，创建一个干净的项目目录并安装核心库。
# 创建项目目录
mkdir aws-mcp-server && cd aws-mcp-server
# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate # Linux/macOS
# venv\Scripts\activate # Windows
# 安装依赖
pip install fastmcp boto3
这里我们安装了三个包：
fastmcp：构建MCP服务器的核心框架。boto3：AWS的官方Python SDK，用于调用DynamoDB和S3等服务。
3.2 编写MCP服务器核心代码
创建文件 aws_mcp_server.py ，我们将逐步填充内容。
import boto3
import json
from datetime import datetime
from typing import Optional
from fastmcp import FastMCP
# 1. 初始化FastMCP服务器实例
# 给服务器起个名字，这会在MCP客户端中显示
mcp = FastMCP("AWS Tools Server")
# 2. 初始化AWS服务客户端
# 注意：这里使用默认的AWS凭证链和区域。
# 在生产部署中（如AgentCore），容器内会通过任务角色（Task Role）获得权限。
# 本地测试时，依赖的是 `aws configure` 设置的凭证。
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
s3_client = boto3.client("s3", region_name="us-east-1")
# 3. 定义第一个工具：查询DynamoDB
@mcp.tool()
def query_dynamodb(table_name: str, key_name: str, key_value: str) -> str:
"""
通过主键查询DynamoDB表中的特定记录。
当用户想要从数据库中查找特定记录、客户数据、订单信息或任何存储在DynamoDB中的结构化数据时，使用此工具。
Args:
table_name: 要查询的DynamoDB表名。
key_name: 主键属性名（分区键）。对于复合主键，目前只支持分区键查询。
key_value: 要查找的主键值。
Returns:
一个JSON格式的字符串。如果找到记录，包含记录详情；如果未找到，返回明确提示；如果出错，返回错误信息。
"""
try:
# 获取表对象
table = dynamodb.Table(table_name)
# 执行GetItem操作
response = table.get_item(Key={key_name: key_value})
# 提取返回的条目
item = response.get("Item")
if not item:
# 未找到记录，返回结构化的“未找到”信息，便于AI理解
return json.dumps({
"success": False,
"found": False,
"message": f"No record found for {key_name} = {key_value} in table '{table_name}'.",
"table": table_name,
"key": {key_name: key_value}
}, default=str)
# 找到记录，返回成功和数据
return json.dumps({
"success": True,
"found": True,
"table": table_name,
"record": item
}, default=str) # default=str 用于处理JSON不支持的日期等类型
except dynamodb.meta.client.exceptions.ResourceNotFoundException:
# 处理表不存在的情况
return json.dumps({
"success": False,
"error": f"DynamoDB table '{table_name}' not found.",
"error_type": "ResourceNotFound"
})
except Exception as e:
# 捕获其他所有异常，避免服务器崩溃
return json.dumps({
"success": False,
"error": str(e),
"error_type": type(e).__name__,
"table": table_name
})
# 4. 定义第二个工具：获取S3桶摘要
@mcp.tool()
def get_s3_summary(bucket_name: str, prefix: str = "") -> str:
"""
列出并总结S3存储桶中的文件。
当用户询问关于文件、文档、日志或存储在S3中的任何数据时，使用此工具。
Args:
bucket_name: 要检查的S3存储桶名称。
prefix: 可选的前缀，用于过滤结果（例如 'reports/' 或 'data/2026/'）。
默认为空字符串，表示桶的根目录。
Returns:
一个JSON格式的字符串，包含文件数量、总大小、文件列表（前20个）等信息。
"""
try:
# 使用分页器处理可能包含大量文件的桶
paginator = s3_client.get_paginator("list_objects_v2")
# 配置分页，限制最大返回项目数，防止返回数据过大
operation_parameters = {
"Bucket": bucket_name,
"Prefix": prefix,
}
# 使用分页迭代器
page_iterator = paginator.paginate(**operation_parameters)
files = []
total_size = 0
file_count = 0
for page in page_iterator:
# S3可能返回没有'Contents'键的页（如空桶或前缀无匹配）
for obj in page.get("Contents", []):
file_count += 1
total_size += obj["Size"]
# 我们只收集前20个文件的详细信息，以节省上下文窗口
if len(files) < 20:
files.append({
"key": obj["Key"],
"size_kb": round(obj["Size"] / 1024, 2),
"last_modified": obj["LastModified"].isoformat()
})
# 即使超过20个，我们也继续计数和累加大小，但不再添加详情
# 构建响应
summary = {
"success": True,
"bucket": bucket_name,
"prefix": prefix if prefix else "(root)",
"file_count": file_count,
"total_size_mb": round(total_size / (1024 * 1024), 2),
"sampled_files": files, # 明确表明这是采样数据
"note": f"Showing {len(files)} sample files out of {file_count} total."
}
# 如果文件太多，添加一个提示
if file_count > 1000:
summary["warning"] = "Bucket contains a large number of files. Consider using more specific prefixes for efficient querying."
return json.dumps(summary, default=str)
except s3_client.exceptions.NoSuchBucket:
return json.dumps({
"success": False,
"error": f"S3 bucket '{bucket_name}' does not exist or you do not have permission to access it.",
"error_type": "NoSuchBucket"
})
except Exception as e:
return json.dumps({
"success": False,
"error": str(e),
"error_type": type(e).__name__,
"bucket": bucket_name
})
# 5. 启动服务器
if __name__ == "__main__":
# 以标准输入输出（stdio）模式运行，这是与Claude Desktop等本地客户端通信的标准方式
mcp.run(transport="stdio")
代码深度解析与避坑指南：
- 工具描述是灵魂 ：注意看两个工具函数下的文档字符串（docstring）。
fastmcp会将这些描述转换为工具定义的一部分。AI模型（如Claude） 完全依赖这些描述来判断何时调用工具 。因此，描述必须清晰、具体，并包含使用场景。例如，“当用户想要从数据库中查找特定记录、客户数据、订单信息...”这样的描述，比简单的“查询数据库”要好得多。 - 错误处理至关重要 ：每个工具都用
try...except包裹，并捕获了特定的服务异常（如ResourceNotFoundException,NoSuchBucket）。 绝对不能让未处理的异常抛出到MCP协议层 ，这会导致整个工具调用链断裂，智能体会收到一个晦涩的错误。我们的策略是始终返回结构化的JSON，包含success字段，让AI能理解操作是成功还是失败，并能将错误信息友好地转述给用户。 - 限制返回数据量 ：在
get_s3_summary中，我们只收集前20个文件的详细信息。这是因为工具返回的所有数据都会塞进AI模型的上下文窗口（Context Window）。一个返回上万行数据的工具调用会瞬间耗尽宝贵的上下文令牌（Tokens），导致后续对话无法进行。 设计MCP工具时，“节制”是一种美德 。优先返回摘要、统计信息或分页数据。 - 使用
default=str：DynamoDB和S3返回的数据中可能包含datetime对象，这不是原生JSON可序列化的。json.dumps(..., default=str)会将所有非序列化对象转换为字符串，避免序列化错误。
3.3 在本地进行测试：连接Claude Desktop
在投入云环境前，先在本地验证MCP服务器是否工作。最方便的方法是使用 Claude Desktop 或 Cursor （如果它们已支持MCP）。
-
首先，运行你的服务器 ：
python aws_mcp_server.py服务器会启动并等待标准输入（stdio）上的连接。先保持它运行。
-
配置Claude Desktop ：
- 打开Claude Desktop应用。
- 进入设置（Settings） -> 开发者（Developer） -> MCP服务器。
- 点击“添加服务器”（Add Server）。
- 给它起个名字，比如
aws-tools。 - 在“命令”（Command）字段中，填入你的Python解释器路径和脚本路径。例如：
（Windows用户可能是/path/to/your/venv/bin/python /path/to/your/aws-mcp-server/aws_mcp_server.pyC:\path\to\venv\Scripts\python.exe C:\path\to\aws_mcp_server.py） - 保存并重启Claude Desktop。
-
进行测试 ：
- 重启后，在新的对话中，你可以直接问：“帮我查一下S3桶
my-test-bucket里有什么文件？”（请替换为你有读取权限的真实桶名）。 - 如果配置正确，Claude应该会识别出
get_s3_summary工具，并在后台调用它。你会在Claude的回复中看到工具调用的结果。 - 同样，测试DynamoDB查询：“在
UserTable表中查找主键user_id等于alice123的记录。”
- 重启后，在新的对话中，你可以直接问：“帮我查一下S3桶
本地测试常见问题 ：
- Claude没有调用工具 ：首先检查Claude Desktop的MCP服务器配置是否正确，服务器进程是否在运行。其次， 仔细检查你的工具描述 。如果描述太模糊，Claude可能无法关联用户问题与工具功能。尝试将问题问得更直接，如“使用AWS工具查看我的S3桶”。
- 权限错误 ：确保你本地
aws configure使用的凭证有足够的S3和DynamoDB权限。可以在终端用aws s3 ls my-test-bucket和aws dynamodb get-item ...命令先验证权限。- 连接错误 ：确保Claude Desktop的命令路径完全正确，特别是虚拟环境下的Python路径。
4. 部署到生产环境：AWS Bedrock AgentCore Runtime
本地测试通过后，我们就可以将其部署到AWS的托管环境中了。我们将使用Bedrock AgentCore Runtime，它负责所有繁琐的运维工作。
4.1 容器化：创建Dockerfile和依赖文件
首先，我们需要将应用打包成Docker镜像。
-
创建
requirements.txt文件 ：fastmcp==0.9.0 boto3==1.35.0固定版本号以确保生产环境的一致性。
-
创建
Dockerfile文件 ：# 使用官方的轻量级Python镜像 FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY aws_mcp_server.py . # AgentCore Runtime 期望MCP服务器运行在8080端口 EXPOSE 8080 # 以HTTP传输模式运行，这是AgentCore要求的 # 注意：这里我们修改了启动命令，使用 `--transport http` CMD ["python", "aws_mcp_server.py", "--transport", "http", "--port", "8080"]关键点 ：本地测试时我们使用
stdio传输，因为是与桌面应用直接通信。在AgentCore Runtime中，服务器需要通过HTTP提供服务，因此启动命令必须指定--transport http。
4.2 构建并推送镜像到Amazon ECR
Amazon Elastic Container Registry (ECR) 是AWS托管的Docker镜像仓库。
-
在AWS控制台或使用CLI创建ECR仓库 ：
# 替换 `123456789` 为你的AWS账号ID，`us-east-1`为你的区域 aws ecr create-repository --repository-name aws-tools-mcp-server --region us-east-1记下返回的
repositoryUri，例如123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server。 -
登录到ECR ：
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 123456789.dkr.ecr.us-east-1.amazonaws.com -
构建并标记镜像 ：
docker build -t aws-tools-mcp-server . docker tag aws-tools-mcp-server:latest 123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest -
推送镜像 ：
docker push 123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest
4.3 在Bedrock AgentCore Runtime中部署MCP服务器
现在，我们将这个镜像部署为托管服务。
-
确保你有正确的IAM权限 ：执行部署的用户或角色需要
bedrock-agentcore:CreateAgentRuntime等权限。最简单的方法是附加AWS托管的AmazonBedrockFullAccess策略（仅用于测试），生产环境请创建更精细的策略。 -
使用Python脚本或AWS CLI创建运行时 ： 创建一个名为
deploy_to_agentcore.py的脚本：import boto3 import time # 初始化客户端，确保区域与ECR镜像和你的资源一致 agentcore_client = boto3.client("bedrock-agentcore", region_name="us-east-1") # 你的ECR镜像URI ECR_IMAGE_URI = "123456789.dkr.ecr.us-east-1.amazonaws.com/aws-tools-mcp-server:latest" def create_mcp_runtime(): """ 在Bedrock AgentCore Runtime中创建一个MCP服务器运行时。 """ try: response = agentcore_client.create_agent_runtime( agentRuntimeName="aws-tools-mcp-server", # 运行时的唯一名称 agentRuntimeArtifact={ "containerConfiguration": { "containerUri": ECR_IMAGE_URI, # 可以在这里定义环境变量、命令等，本例使用Dockerfile中的默认CMD } }, networkConfiguration={ "networkMode": "PUBLIC" # 或 "VPC"，如果你的服务需要访问VPC内资源 }, # 可选：指定计算配置（CPU/内存） # computeConfiguration={ # 'cpu': '1024', # 'memoryInMB': '2048' # } ) runtime_arn = response["agentRuntimeArn"] runtime_endpoint = response["agentRuntimeEndpoint"] runtime_id = response["agentRuntimeId"] print(f"✅ AgentCore Runtime 创建成功！") print(f" ARN: {runtime_arn}") print(f" ID: {runtime_id}") print(f" Endpoint: {runtime_endpoint}") print(f"\n⚠️ 请注意：运行时启动可能需要几分钟。请使用 describe_agent_runtime 检查状态。") return runtime_endpoint except Exception as e: print(f"❌ 创建运行时失败: {e}") return None if __name__ == "__main__": endpoint = create_mcp_runtime() if endpoint: # 你可以在这里添加一个轮询，等待运行时状态变为 'ACTIVE' print(f"\n部署完成后，你的MCP服务器端点将是: {endpoint}") -
运行部署脚本并等待激活 ：
python deploy_to_agentcore.py执行后，脚本会返回一个运行时ARN和一个端点URL（如
https://xxx.bedrock-agentcore.us-east-1.amazonaws.com）。 这个端点URL就是你的MCP服务器的访问地址 ，稍后需要提供给Bedrock Agent。运行时从创建到状态变为
ACTIVE可能需要几分钟。你可以使用以下命令检查状态：aws bedrock-agentcore describe-agent-runtime --agent-runtime-id <你的RuntimeId> --region us-east-1查看返回的
status字段，等待其变为ACTIVE。
4.4 将MCP服务器连接到Bedrock Agent
最后一步，创建一个Bedrock Agent（或使用现有的），并配置它使用我们刚部署的MCP服务器。
这里我们直接编写一个与Bedrock Runtime对话的Python脚本，它模拟了一个集成了MCP工具的智能体工作流。
import boto3
import json
from typing import Dict, Any
# 初始化Bedrock Runtime客户端
bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")
# 替换为你的AgentCore Runtime端点
MCP_SERVER_ENDPOINT = "https://your-unique-id.bedrock-agentcore.us-east-1.amazonaws.com"
def run_conversation_with_tools(user_query: str) -> str:
"""
执行一个与Bedrock模型的对话循环，该模型可以使用我们定义的MCP工具。
"""
# 1. 定义系统提示词，指导模型使用工具
system_prompt = f"""
你是一个专业的AWS运维助手，可以访问用户的AWS资源来获取实时数据。
你连接了一个MCP服务器，它提供了以下工具：
1. `query_dynamodb`: 通过主键查询DynamoDB表中的记录。
2. `get_s3_summary`: 列出并总结S3存储桶中的文件。
当用户询问关于S3文件、DynamoDB记录或其他相关AWS数据的问题时，你必须使用相应的工具来获取真实数据。
绝对不要猜测或编造数据。如果工具返回错误，请将错误信息友好地告知用户。
MCP服务器地址: {MCP_SERVER_ENDPOINT}
"""
# 2. 初始化对话历史
messages = [
{
"role": "user",
"content": [{"text": user_query}]
}
]
# 3. 定义工具配置（告诉模型有哪些工具可用）
# 注意：这里的工具定义（名称、描述、输入模式）必须与MCP服务器提供的完全匹配。
# FastMCP会自动生成这些模式，但在这里我们需要手动声明给Bedrock API。
tool_config = {
"tools": [
{
"toolSpec": {
"name": "query_dynamodb",
"description": "Query a DynamoDB table by primary key. Use this when the user wants to look up specific records from a database.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"table_name": {"type": "string"},
"key_name": {"type": "string"},
"key_value": {"type": "string"}
},
"required": ["table_name", "key_name", "key_value"]
}
}
}
},
{
"toolSpec": {
"name": "get_s3_summary",
"description": "List and summarize files in an S3 bucket. Use this when the user asks about files, documents, or data stored in S3.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"bucket_name": {"type": "string"},
"prefix": {"type": "string"}
},
"required": ["bucket_name"]
}
}
}
}
]
}
# 4. 开始对话循环（限制最大轮次防止无限循环）
max_turns = 10
for turn in range(max_turns):
try:
# 调用Bedrock Converse API
response = bedrock_runtime.converse(
modelId="anthropic.claude-3-5-sonnet-20241022-v2:0", # 使用最新的Claude 3.5 Sonnet
system=[{"text": system_prompt}],
messages=messages,
toolConfig=tool_config
)
# 解析响应
stop_reason = response['stopReason']
output_message = response['output']['message']
# 将模型的回复加入历史
messages.append(output_message)
# 情况A: 模型决定结束本轮对话
if stop_reason == 'end_turn':
# 提取模型的最终文本回复并返回
for content_block in output_message['content']:
if 'text' in content_block:
return content_block['text']
return "模型结束了对话，但未返回文本。"
# 情况B: 模型决定使用工具
elif stop_reason == 'tool_use':
tool_results_content = []
print(f"\n[Turn {turn+1}] 模型请求使用工具...")
for content_block in output_message['content']:
if 'toolUse' not in content_block:
continue
tool_use_block = content_block['toolUse']
tool_name = tool_use_block['name']
tool_input = tool_use_block['input']
tool_use_id = tool_use_block['toolUseId']
print(f" 工具: {tool_name}, 输入: {tool_input}")
# 这里是一个关键点！
# 在实际的、完整的Bedrock Agent with MCP集成中，当模型决定使用工具时，
# Bedrock服务会自动将请求路由到你在Agent配置中指定的MCP服务器端点。
# 你不需要像下面这样在代码中手动调用函数。
# 这个手动调用仅用于演示逻辑，或者在你完全自己控制对话流时使用。
# 模拟工具调用结果（在实际MCP集成中，这部分由Bedrock服务自动完成）
# 我们这里只是打印信息，并构造一个模拟的成功响应。
simulated_result = json.dumps({
"success": True,
"note": f"[模拟] 工具 '{tool_name}' 被调用。在实际Bedrock Agent MCP集成中，此调用会自动路由到 {MCP_SERVER_ENDPOINT}",
"input_received": tool_input
})
# 构造工具调用结果消息块
tool_results_content.append({
"toolResult": {
"toolUseId": tool_use_id,
"content": [{"text": simulated_result}],
# 可以添加 status 字段，如 "success" 或 "error"
# "status": "success"
}
})
# 将工具调用结果作为用户消息加入历史，让模型进行下一轮推理
if tool_results_content:
messages.append({
"role": "user",
"content": tool_results_content
})
else:
# 如果没有工具调用结果，跳出循环
break
else:
# 其他停止原因（如长度限制）
return f"对话因 '{stop_reason}' 而停止。"
except Exception as e:
return f"在与Bedrock API交互时发生错误: {e}"
return f"已达到最大对话轮次 ({max_turns})，未完成请求。"
# 5. 测试这个集成
if __name__ == "__main__":
# 测试查询
test_queries = [
"我的S3桶 'my-data-bucket' 里有多少个文件？",
# "在DynamoDB表 'Users' 中查找主键 'user_id' 为 'test_user_1' 的记录。",
# "帮我总结一下 'logs-bucket' 桶中 '2024-12/' 前缀下的文件。"
]
for query in test_queries:
print(f"\n{'='*60}")
print(f"用户提问: {query}")
print(f"{'='*60}")
answer = run_conversation_with_tools(query)
print(f"助手回答:\n{answer}\n")
关键说明与下一步 ： 上面的脚本演示了Bedrock Converse API如何与工具交互。然而，要完成 真正的、自动化的MCP集成 ，你通常需要在 AWS Bedrock 控制台 中操作：
- 进入 Bedrock 控制台 -> 代理 (Agents) 。
- 创建或编辑一个代理。
- 在 工具 (Tools) 部分，选择 添加工具 (Add tool) -> 通过MCP服务器 (Via MCP server) 。
- 输入你的 AgentCore Runtime端点URL （即
MCP_SERVER_ENDPOINT）。 - 保存代理。
完成此配置后，当你通过该代理与模型对话时，Bedrock服务会自动处理MCP协议的通信：发现工具、路由工具调用、返回结果。你无需再在应用代码中手动处理工具调用逻辑，模型会根据你的问题自动选择并使用工具。
5. 生产环境进阶指南与避坑实录
将MCP服务器部署上线只是第一步，要保证其稳定、高效、安全地运行，还需要注意以下几个关键点，这些都是我从实际项目中踩坑总结出来的经验。
5.1 工具描述的“艺术”：让AI准确理解你的意图
这是导致MCP集成失败的头号原因。模型的工具调用决策几乎完全基于你提供的工具描述（ description ）和参数名。
-
反面教材 ：
@mcp.tool() def get_data(table: str, key: str) -> str: """从表获取数据。""" # ...这种描述过于模糊。模型无法准确判断何时该调用它。“数据”是什么？“表”是什么？用户问“我的订单状态”时会调用它吗？很可能不会。
-
最佳实践 ：
@mcp.tool() def query_dynamodb(table_name: str, key_name: str, key_value: str) -> str: """ 通过主键查询DynamoDB表中的特定记录。 当用户想要从数据库中查找特定记录、客户数据、订单信息、产品详情或任何存储在DynamoDB中的结构化数据时，使用此工具。 用户的问题中通常会包含表名、ID、编号等查找条件。 """ # ...- 功能明确 ：开头就说明是“通过主键查询”。
- 场景枚举 ：列出典型的使用场景（“查找客户数据、订单信息”）。
- 参数名自解释 ：
table_name,key_name,key_value比table,key清晰得多。 - 提示用户提问模式 ：说明用户通常会如何提问，帮助模型建立关联。
实操心得 ：写好描述后，用各种同义但非直接的问法去测试。例如，对于查询S3的工具，不仅要测试“列出我的桶里文件”，还要测试“我的文档存在S3里了吗？”、“备份文件有多少？”。观察模型是否能正确触发工具。
5.2 性能与成本：控制上下文窗口的消耗
大型语言模型的上下文窗口是宝贵且有限的资源。工具返回的所有数据都会占用这个窗口。
- 问题 ：一个
get_s3_summary工具，如果桶里有10万个文件，即使你只返回文件名列表，也可能瞬间消耗数万tokens，导致后续对话无法进行或API调用成本激增。 - 解决方案 ：
- 强制分页/限制 ：像我们的示例一样，始终对返回的数据量施加硬性限制（例如，最多返回20个条目）。在工具描述中说明这一点。
- 返回摘要，而非原始数据 ：计算并返回总数、总大小、最近修改时间等元数据，而不是完整的列表。如果需要详情，可以设计另一个支持分页参数的工具（如
get_s3_files(bucket, prefix, limit, offset)）。 - 使用
note字段 ：在返回的JSON中明确告知用户“已采样前N条，共M条”，管理其预期。 - 考虑成本 ：每次工具调用和结果返回都会产生Bedrock API的token费用。设计工具时，要像设计数据库查询一样，思考如何用最少的token返回最有价值的信息。
5.3 健壮性：全面的错误处理与状态管理
生产环境中，一切皆可能出错：网络超时、权限不足、资源不存在、输入无效。
- 必须捕获所有异常 ：每个工具函数都必须有顶层的
try...except，确保任何错误都不会导致MCP服务器进程崩溃。 - 返回结构化的错误信息 ：不要只返回错误字符串。返回一个包含
success: false、error_type和可读message的JSON对象。这允许AI模型理解错误性质，并可能尝试其他方法或向用户给出更具体的指导。try: # ... 业务逻辑 return json.dumps({"success": True, "data": result}) except ClientError as e: error_code = e.response['Error']['Code'] if error_code == 'AccessDeniedException': return json.dumps({ "success": False, "error_type": "PermissionDenied", "message": "I do not have permission to perform this action. Please check the IAM permissions for this MCP server." }) else: return json.dumps({"success": False, "error_type": "AWSClientError", "message": str(e)}) except Exception as e: return json.dumps({"success": False, "error_type": "UnexpectedError", "message": f"An internal error occurred: {e}"}) - 验证输入 ：在业务逻辑开始前，验证输入参数。例如，检查
table_name是否不为空，key_value是否符合预期格式。提前返回清晰的错误，比在深层AWS调用中失败更好。
5.4 安全性与权限：最小权限原则与隔离
MCP服务器本质上是一个拥有AWS API调用权限的代理。安全至关重要。
- IAM角色（Task Role） ：在Bedrock AgentCore Runtime中，为你的运行时分配一个IAM角色。这个角色应遵循 最小权限原则 ，只授予该MCP服务器所需的具体操作权限（如
s3:ListBucket、dynamodb:GetItem）。切勿使用管理员权限。 - 输入净化与校验 ：警惕通过工具参数传入的潜在恶意输入。虽然MCP客户端（模型）通常是可信的，但仍应进行基础校验，防止意外的目录遍历（如S3前缀中的
../）或SQL注入（虽然DynamoDB不涉及SQL，但也要小心NoSQL注入模式）。 - 网络隔离 ：在
CreateAgentRuntime时，networkConfiguration可以选择VPC模式。如果你的MCP服务器需要访问VPC内的资源（如私有子网中的RDS数据库），务必将其部署在VPC内，并通过安全组严格控制入站和出站流量。对于仅需访问公有API（S3, DynamoDB公共端点）的服务，PUBLIC模式即可。 - 会话隔离 ：Bedrock AgentCore Runtime的一个巨大优势是它为每个会话提供独立的微VM。这意味着用户A的工具调用状态（例如，一个长时间运行的查询上下文）不会与用户B的混淆。在设计有状态工具时，可以依赖这个特性。
5.5 可观测性：日志、监控与调试
当工具调用失败或行为异常时，你需要有办法排查。
- 结构化日志 ：在工具函数中使用
print()或logging模块输出结构化日志（JSON格式）。记录工具调用开始、参数、成功/失败、耗时。这些日志会自动集成到Amazon CloudWatch Logs中。import logging logger = logging.getLogger(__name__) @mcp.tool() def my_tool(param): logger.info(json.dumps({"event": "tool_invoked", "tool": "my_tool", "param": param})) try: result = do_work(param) logger.info(json.dumps({"event": "tool_succeeded", "tool": "my_tool", "duration_ms": duration})) return result except Exception as e: logger.error(json.dumps({"event": "tool_failed", "tool": "my_tool", "error": str(e)})) return error_response - CloudWatch Metrics ：你可以使用
boto3向CloudWatch发送自定义指标，例如工具调用次数、平均延迟、错误率。这对于监控服务健康度和设置警报至关重要。 - 在工具响应中包含调试信息（仅开发环境） ：在开发阶段，可以在返回的JSON中添加一个
_debug字段，包含内部ID或简化的日志片段，帮助前端或测试人员定位问题。生产环境应移除或限制此功能。
6. 未来展望：有状态MCP与更复杂的架构
AWS在2026年初为AgentCore Runtime引入了 有状态MCP服务器 的支持，这打开了新的大门。在此之前，每次工具调用都是无状态的，服务器无法记住之前的交互。
- 有状态会话 ：现在，通过
Mcp-Session-Id请求头，服务器可以识别属于同一用户会话的多次调用。这使得以下功能成为可能：- 多轮交互（Elicitation） ：工具执行过程中，可以主动要求用户提供更多信息。例如，一个部署工具可以问：“您要部署到生产环境还是预发布环境？”
- 内容生成（Sampling） ：工具可以请求AI模型在工具执行流程中生成内容。例如，一个代码分析工具可以请求模型对一段查询结果进行总结。
- 长时运行操作 ：对于需要几分钟甚至几小时的任务（如训练机器学习模型、导出大量数据），服务器可以启动任务后立即返回一个“任务已提交”的响应，然后在后台通过会话ID更新状态，而无需保持最初的请求连接。
基于当前的基础，你可以探索更强大的架构：
- 工具生态扩展 ：为你的MCP服务器添加更多AWS工具：
query_cloudwatch_logs（查询日志）、invoke_lambda（触发函数）、describe_ec2_instances（查看实例状态）、start_step_function（启动工作流）。逐渐构建一个覆盖你常用运维操作的AI助手。 - 多代理与编排 ：设计一个“协调员”代理，它负责理解用户的高层目标，然后调用多个专门的“子代理”，每个子代理连接到自己专用的MCP服务器（如数据库专家、日志分析专家、部署专家）。这符合AI领域的“心智社会”理念。
- AgentCore Gateway ：与其在每个代理中硬编码MCP服务器地址，不如使用AgentCore Gateway作为工具的中心注册表。将你的MCP服务器注册到Gateway，然后任何代理都可以通过查询Gateway来发现和使用这些工具，实现更好的解耦和复用。
从我个人的实践来看，MCP与Bedrock AgentCore的结合，正在将AI智能体从“玩具”变为真正的“生产级助手”。它解决了工具集成的标准化和运维的复杂性两大痛点。虽然初期在工具描述设计和错误处理上需要投入精力，但一旦搭建好这个管道，为AI添加新能力就变成了简单的Python函数开发，其回报是巨大的。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
6
0- 0
已为社区贡献8条内容
所有评论(0)