MCP 协议实战:用 Amazon Bedrock 让 AI Agent 安全调用云服务的完整方案
MCP 做的事情是把 Agent 的工具调用标准化。不用每个工具单独写适配代码了,装一个 Server 就获得一组工具。Bedrock Agent 原生支持 MCPIAM 提供了双层权限控制(Agent Role + Server Role)CloudTrail 记录完整调用链路VPC Endpoint 保证网络安全如果你的 Agent 需要接入多个云服务或内部 API,MCP 是目前比较优雅的标
三个月前有人问我:「Agent 怎么调外部 API 比较优雅?」
我当时的回答是写 function calling,每个工具手动定义 schema。能用,但维护成本高——加一个工具就要改代码、改 schema、改权限配置。
后来接触了 MCP(Model Context Protocol),思路一下打开了。
MCP 解决了什么问题
Agent 要干活,就得调工兴——读文件、查数据库、发通知、调 API。传统做法是每个工具单独接入:
Agent → 自定义 function → S3 API
Agent → 自定义 function → DynamoDB API
Agent → 自定义 function → SNS API
Agent → 自定义 function → 自建服务 API
每个 function 都要写:参数定义、鉴权逻辑、错误处理、重试机制。工具一多就是大量重复劳动。
MCP 把这件事标准化了。它定义了一个协议:工具提供方按规范暴露能力,模型/Agent按规范调用。中间走统一的 JSON-RPC 通道。
Agent → MCP Client → MCP Server (S3)
→ MCP Server (DynamoDB)
→ MCP Server (SNS)
→ MCP Server (自建服务)
好处:
- 接入成本低:装一个 MCP Server,Agent 自动获得对应工具
- 标准化:所有工具遵循同一协议,不用每个单独适配
- 可组合:需要什么能力装什么 Server,热插拔
- 安全可控:在 Server 层面做权限限制,Agent 只能调用 Server 暴露的方法
MCP 的核心概念
简单过一下:
Server 和 Client
- MCP Server:提供工具的一方。比如一个 S3 MCP Server 暴露了
list_objects、get_object、put_object三个工具 - MCP Client:调用工具的一方。通常集成在 Agent 或 IDE 里
Transport
Client 和 Server 之间的通信通道。支持:
- stdio:标准输入输出(本地进程)
- HTTP + SSE:网络通信(远程 Server)
Tools, Resources, Prompts
Server 可以暴露三种东西:
- Tools:可执行的操作(读文件、查数据、发请求)
- Resources:只读数据(配置文件、数据库 schema)
- Prompts:预设的 prompt 模板
Bedrock + MCP:亚马逊云科技的集成方案
Amazon Bedrock 原生支持 MCP。你可以在 Bedrock Agent 中接入 MCP Server,让 Agent 通过 MCP 协议调用各种云服务。
架构
用户请求
↓
Bedrock Agent
↓
MCP Client(内置)
↓
┌─────────────┐ ┌─────────────┐ ┌──────────────┐
│ MCP Server │ │ MCP Server │ │ MCP Server │
│ (S3) │ │ (DynamoDB) │ │ (自建 API) │
└─────────────┘ └─────────────┘ └──────────────┘
实战:给 Bedrock Agent 接一个 S3 MCP Server
第 1 步:创建 MCP Server(Python 示例)
from mcp.server import Server
from mcp.types import Tool, TextContent
import boto3
import json
app = Server("s3-mcp-server")
s3 = boto3.client('s3', region_name='cn-northwest-1')
@app.tool()
async def list_files(bucket: str, prefix: str = "") -> str:
"""列出 S3 存储桶中指定前缀下的文件"""
response = s3.list_objects_v2(
Bucket=bucket,
Prefix=prefix,
MaxKeys=50
)
files = [obj['Key'] for obj in response.get('Contents', [])]
return json.dumps(files, ensure_ascii=False)
@app.tool()
async def read_file(bucket: str, key: str) -> str:
"""读取 S3 存储桶中的文本文件内容"""
response = s3.get_object(Bucket=bucket, Key=key)
content = response['Body'].read().decode('utf-8')
# 限制返回长度,防止占满 context window
if len(content) > 10000:
content = content[:10000] + "\n...(内容已截断)"
return content
@app.tool()
async def write_file(bucket: str, key: str, content: str) -> str:
"""写入文本内容到 S3 存储桶"""
s3.put_object(
Bucket=bucket,
Key=key,
Body=content.encode('utf-8'),
ContentType='text/plain'
)
return f"已写入: s3://{bucket}/{key}"
if __name__ == "__main__":
import asyncio
from mcp.server.stdio import stdio_server
asyncio.run(stdio_server(app))
第 2 步:安全配置
MCP Server 跑在 EC2 上,给它配一个最小权限的 IAM Role:
{
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:GetObject",
"s3:PutObject",
"s3:ListBucket"
],
"Resource": [
"arn:aws-cn:s3:::my-agent-bucket",
"arn:aws-cn:s3:::my-agent-bucket/*"
]
}
]
}
注意:
- 只允许访问特定 bucket
- 只给了 Get/Put/List,没给 Delete
- MCP Server 的权限就是 Agent 通过这个 Server 能做的事的上限
第 3 步:在 Bedrock Agent 中配置 MCP Server
import boto3
bedrock_agent = boto3.client('bedrock-agent', region_name='cn-northwest-1')
response = bedrock_agent.create_agent(
agentName='my-s3-agent',
foundationModel='anthropic.claude-3-sonnet-20240229-v1:0',
instruction="""你是一个文件管理助手。
用户可以让你查看、读取或写入 S3 存储桶中的文件。
所有操作限定在 my-agent-bucket 存储桶内。
不要访问其他存储桶。""",
agentResourceRoleArn='arn:aws-cn:iam::123456789012:role/BedrockAgentRole'
)
MCP 的安全考量
MCP 让 Agent 接入工具变得容易了,但也带来了安全问题。几个要点:
1. Server 层面限权
MCP Server 决定了 Agent 能调什么工具、能传什么参数。安全做在 Server 里:
@app.tool()
async def read_file(bucket: str, key: str) -> str:
# 白名单检查
allowed_buckets = ["my-agent-bucket"]
if bucket not in allowed_buckets:
return "错误:无权访问该存储桶"
# 路径检查,防止目录遍历
if ".." in key or key.startswith("/"):
return "错误:非法文件路径"
# 正常执行
response = s3.get_object(Bucket=bucket, Key=key)
return response['Body'].read().decode('utf-8')
2. IAM 双重保险
MCP Server 的 IAM Role 是第一道限制,Bedrock Agent 的 IAM Role 是第二道。两个取交集。
3. 日志和审计
MCP Server 的每次调用都应该记日志:
import logging
logger = logging.getLogger("mcp-s3")
@app.tool()
async def read_file(bucket: str, key: str) -> str:
logger.info(f"read_file called: bucket={bucket}, key={key}")
# ...
配合 CloudTrail,形成完整的调用链路追踪。
4. 传输安全
生产环境用 HTTP + TLS 传输,不要用 stdio(stdio 只适合本地开发)。
哪些场景适合 MCP
| 场景 | 适合 MCP 吗 | 原因 |
|---|---|---|
| Agent 需要读写 S3 | ✅ | 标准化接入,权限可控 |
| Agent 需要查 DynamoDB | ✅ | Server 层面限制可查的表和字段 |
| Agent 需要调内部 API | ✅ | 把内部 API 包装成 MCP Server |
| 简单的单工具调用 | 不一定 | 直接 function calling 可能更简单 |
| 实时流式数据 | ❌ | MCP 是请求/响应模式,不适合流 |
总结
MCP 做的事情是把 Agent 的工具调用标准化。不用每个工具单独写适配代码了,装一个 Server 就获得一组工具。
跟 Amazon Bedrock 配合的好处是:
- Bedrock Agent 原生支持 MCP
- IAM 提供了双层权限控制(Agent Role + Server Role)
- CloudTrail 记录完整调用链路
- VPC Endpoint 保证网络安全
如果你的 Agent 需要接入多个云服务或内部 API,MCP 是目前比较优雅的标准化方案。
🔗 Amazon Bedrock:https://aws.amazon.com/cn/bedrock/
🔗 Amazon Bedrock AgentCore:https://aws.amazon.com/cn/bedrock/agentcore/
🔗 IAM 文档:https://docs.aws.amazon.com/iam/
🔗 CloudTrail:https://aws.amazon.com/cn/cloudtrail/
更多推荐

所有评论(0)