三个月前有人问我:「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 (自建服务)

好处:

  1. 接入成本低:装一个 MCP Server,Agent 自动获得对应工具
  2. 标准化:所有工具遵循同一协议,不用每个单独适配
  3. 可组合:需要什么能力装什么 Server,热插拔
  4. 安全可控:在 Server 层面做权限限制,Agent 只能调用 Server 暴露的方法

MCP 的核心概念

简单过一下:

Server 和 Client

  • MCP Server:提供工具的一方。比如一个 S3 MCP Server 暴露了 list_objectsget_objectput_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/

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐