模型能"说"要调用工具，但谁来真正执行代码、返回结果？Amazon Bedrock 近期推出的 Programmatic Tool Calling（PTC）把这个问题推到了开发者面前：你需要一个沙箱来安全地跑代码，再把输出喂回模型。选哪种沙箱方案，直接决定了你的控制粒度、运维成本和开发体验。
下面拆解三种实现方式——自建 ECS Docker 沙箱、Bedrock AgentCore Code Interpreter 托管方案、以及 Anthropic SDK 代理路径——并给出可落地的配置示例。
为什么 PTC 不等于普通 Tool Calling
传统 Tool Calling 是模型输出一个 JSON 结构，你的应用层解析后调用外部 API，再把结果拼回对话。PTC 的区别在于：模型请求的是执行一段代码（通常是 Python），而不是调一个 REST 接口。这意味着你需要一个能安全运行任意代码的执行环境，而不是一个 API 网关。
核心挑战有三：
- 安全性：用户间接提供的代码不能逃逸沙箱、访问宿主机资源。
- 延迟：代码执行结果要快速回流，否则多轮对话体验会崩。
- 状态管理：执行环境是否跨轮次保持（比如保留已导入的库和变量）。
三种方案分别在这些维度上做了不同取舍。
方案一：ECS 上的自托管 Docker 沙箱
把代码执行塞进 ECS Task，每个请求启动一个隔离容器，执行完销毁。你拥有完全控制权——镜像内容、网络策略、资源限额全由你定。
代价是运维负担：你要自己处理容器生命周期、日志收集、弹性伸缩，还要确保容器间网络隔离。
下面是一个最小化的 ECS Task Definition 和配套的执行代理微服务骨架：
# task-definition.yaml — ECS Fargate 沙箱任务定义
awsEcsTaskDefinition:
family: ptc-sandbox
networkMode: awsvpc
requiresCompatibilities:
- FARGATE
cpu: "512"
memory: "1024"
containerDefinitions:
- name: sandbox-runner
image: "123456789012.dkr.ecr.us-east-1.amazonaws.com/ptc-sandbox:latest"
essential: true
resourceRequirements:
- type: GPU
value: "0" # 不需要 GPU，纯 CPU 执行
environment:
- name: MAX_EXEC_SECONDS
value: "30" # 执行超时上限
- name: MEMORY_LIMIT_MB
value: "512"
linuxParameters:
initProcessEnabled: true
logConfiguration:
logDriver: awslogs
options:
awslogs-group: /ecs/ptc-sandbox
awslogs-region: us-east-1
awslogs-stream-prefix: sandbox
沙箱镜像的 Dockerfile 关键片段——只装必要的库，不装网络工具：
FROM python:3.11-slim
# 只安装计算类库，不装 requests/httpx 等网络库
RUN pip install --no-cache-dir numpy pandas sympy
# 切到非 root 用户
RUN useradd --create-home sandbox
USER sandbox
WORKDIR /home/sandbox
# 执行入口：从 stdin 读代码，写结果到 stdout
COPY run_code.py /home/sandbox/run_code.py
ENTRYPOINT ["python", "/home/sandbox/run_code.py"]
run_code.py 的核心逻辑——受限执行、超时保护：
import sys
import json
import signal
import resource
MAX_MEMORY = 512 * 1024 * 1024 # 512 MB
def limit_memory():
resource.setrlimit(resource.RLIMIT_AS, (MAX_MEMORY, MAX_MEMORY))
def timeout_handler(signum, frame):
raise TimeoutError("Execution exceeded time limit")
def run(code: str, timeout_sec: int = 30):
limit_memory()
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_sec)
try:
# 只暴露安全内建函数
safe_globals = {"__builtins__": {
"print": print, "len": len, "range": range,
"int": int, "float": float, "str": str,
"list": list, "dict": dict, "set": set,
"True": True, "False": False, "None": None,
}}
# 注入常用库
safe_globals.update(
__import__("numpy").__dict__,
__import__("pandas").__dict__,
)
exec(code, safe_globals)
signal.alarm(0)
except TimeoutError:
print(json.dumps({"error": "timeout", "status": "failed"}))
except Exception as e:
print(json.dumps({"error": str(e), "status": "failed"}))
finally:
signal.alarm(0)
if __name__ == "__main__":
code_input = sys.stdin.read()
run(code_input)
调用侧的编排逻辑：Bedrock 返回 tool_use 后，你的应用启动 ECS Task，把代码通过 stdin 注入，读 stdout 结果，再调用 Bedrock 的 tool_result 接口回传。整个过程需要你写一个调度器来管理 Task 的启停和结果轮询。
适合谁：对安全合规有硬性要求、需要自定义镜像内容（比如装了公司内部计算库）、愿意承担运维成本的团队。
方案二：Bedrock AgentCore Code Interpreter——托管沙箱
Amazon Bedrock AgentCore Code Interpreter 是 AWS 提供的托管代码执行环境。你不需要管容器、不需要管镜像，只需要在 Agent 配置里声明启用 Code Interpreter，Bedrock 自动处理沙箱创建、代码执行和结果回传。
配置方式在 Agent 定义中添加 codeInterpreter capability：
import boto3
client = boto3.client("bedrock-agent")
response = client.create_agent(
agentName="ptc-agent-with-interpreter",
agentResourceRoleArn="arn:aws:iam::123456789012:role/BedrockAgentRole",
idleSessionTTLInSeconds=600,
instruction="You are a data analysis assistant. When asked to compute, generate and run Python code using the code interpreter.",
capabilities=[
{
"codeInterpreter": {
"enabled": True
}
}
],
)
agent_id = response["agent"]["agentId"]
print(f"Agent created: {agent_id}")
调用侧更简单——你只需要正常发起对话，模型自动决定是否调用 Code Interpreter，执行结果由托管层注入回对话流：
import boto3
client = boto3.client("bedrock-agent-runtime")
response = client.invoke_agent(
agentId="YOUR_AGENT_ID",
agentAliasId="TSTALIASID",
sessionId="session-001",
inputText="计算 1 到 1000 的所有质数之和，并给出前 10 个质数",
)
# 流式读取事件，Code Interpreter 的执行结果已自动嵌入
for event in response["completion"]:
if "chunk" in event:
print(event["chunk"]["bytes"].decode())
取舍：你失去了对执行环境的完全控制——不能自定义预装库、不能限制内存到精确值、不能审计容器日志的每一行。换来的是零运维和更快的上线速度。
适合谁：快速验证 PTC 能力、不想运维基础设施、对执行环境定制需求不高的场景。
方案三：Anthropic SDK 代理路径
有些团队已经深度使用 Anthropic 的 Python SDK（anthropic 包），习惯了 client.beta.tools.use() 的开发范式。直接迁移到 Bedrock 的 SDK 意味着改写大量调用代码。
第三种方案是在中间架一个轻量代理：你的应用继续用 Anthropic SDK 的接口风格发请求，代理层把请求翻译成 Bedrock 的 PTC 格式，再把 Bedrock 的沙箱执行结果翻译回 Anthropic SDK 的 tool_result 格式。
代理的核心转换逻辑示意：
from fastapi import FastAPI, Request
import boto3
import json
app = FastAPI()
bedrock = boto3.client("bedrock-runtime")
ANTHROPIC_TO_BEDROCK_MODEL = {
"claude-3-5-sonnet-20241022": "anthropic.claude-3-5-sonnet-20241022-v2:0",
}
@app.post("/v1/messages")
async def proxy_messages(request: Request):
body = await request.json()
# 把 Anthropic SDK 的 tool 定义翻译成 Bedrock toolConfig
anthropic_tools = body.get("tools", [])
bedrock_tools = []
for t in anthropic_tools:
if t["type"] == "custom" and t["name"] == "code_interpreter":
bedrock_tools.append({
"toolSpec": {
"name": t["name"],
"description": t["description"],
"inputSchema": {
"json": t["input_schema"]
}
}
})
model_id = ANTHROPIC_TO_BEDROCK_MODEL.get(
body["model"], "anthropic.claude-3-5-sonnet-20241022-v2:0"
)
bedrock_response = bedrock.converse(
modelId=model_id,
messages=translate_messages(body["messages"]),
toolConfig={"tools": bedrock_tools} if bedrock_tools else None,
inferenceConfig={
"maxTokens": body.get("max_tokens", 4096),
"temperature": body.get("temperature", 0.0),
},
)
# 把 Bedrock 响应翻译回 Anthropic SDK 格式
return translate_response(bedrock_response)
def translate_messages(anthropic_msgs):
"""把 Anthropic 消息格式转为 Bedrock 消息格式"""
bedrock_msgs = []
for msg in anthropic_msgs:
role = msg["role"]
content_blocks = []
for c in msg.get("content", []):
if c["type"] == "text":
content_blocks.append({"text": c["text"]})
elif c["type"] == "tool_use":
content_blocks.append({
"toolUse": {
"toolUseId": c["id"],
"name": c["name"],
"input": c["input"],
}
})
elif c["type"] == "tool_result":
content_blocks.append({
"toolResult": {
"toolUseId": c["tool_use_id"],
"content": [{"text": c["content"]}],
"status": "success" if not c.get("is_error") else "error",
}
})
bedrock_msgs.append({"role": role, "content": content_blocks})
return bedrock_msgs
def translate_response(bedrock_resp):
"""把 Bedrock converse 响应翻译回 Anthropic Messages API 格式"""
output = bedrock_resp["output"]["message"]
content = []
for block in output["content"]:
if "text" in block:
content.append({"type": "text", "text": block["text"]})
elif "toolUse" in block:
content.append({
"type": "tool_use",
"id": block["toolUse"]["toolUseId"],
"name": block["toolUse"]["name"],
"input": block["toolUse"]["input"],
})
return {
"id": bedrock_resp["ResponseMetadata"]["RequestId"],
"type": "message",
"role": output["role"],
"content": content,
"model": bedrock_resp["modelId"],
"stop_reason": map_stop_reason(bedrock_resp["stopReason"]),
"usage": {
"input_tokens": bedrock_resp["usage"]["inputTokens"],
"output_tokens": bedrock_resp["usage"]["outputTokens"],
},
}
def map_stop_reason(reason):
mapping = {
"tool_use": "tool_use",
"end_turn": "end_turn",
"max_tokens": "max_tokens",
}
return mapping.get(reason, reason)
部署这个代理到 Lambda 或 ECS 后，你的现有代码只需改一行——把 base_url 指向代理：
from anthropic import Anthropic
# 原来的调用：client = Anthropic()
# 改为指向代理：
client = Anthropic(base_url="https://your-proxy.example.com/v1")
response = client.beta.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=4096,
tools=[{
"type": "custom",
"name": "code_interpreter",
"description": "Execute Python code in a sandbox",
"input_schema": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to execute"}
},
"required": ["code"],
},
}],
messages=[{"role": "user", "content": "用 Python 计算 Fibonacci 数列前 20 项"}],
)
取舍：代理层引入了额外延迟（一次 HTTP 转发 + 格式转换），而且你需要自己维护翻译逻辑的完整性——Anthropic 和 Bedrock 的消息格式差异会随版本演进扩大。但团队不需要重写调用代码，迁移成本最低。
适合谁：已有 Anthropic SDK 集成、想渐进迁移到 Bedrock、暂时不愿大规模改写调用层的团队。
选型对照
| 维度 | ECS 自托管 | AgentCore Code Interpreter | Anthropic SDK 代理 |
|---|---|---|---|
| 控制粒度 | 完全自定义镜像、资源限额 | 托管层决定，有限配置 | 取决于后端选哪种沙箱 |
| 运维成本 | 高（容器管理、伸缩、日志） | 低（AWS 全托管） | 中（维护代理服务） |
| 上线速度 | 慢（需构建镜像+调度器） | 快（Agent 配置即启用） | 中（需部署代理+调试翻译） |
| 安全审计 | 完全可控，可逐行审计 | 依赖 AWS 审计能力 | 代理层可加日志，沙箱取决于后端 |
| SDK 体验 | Bedrock SDK 原生 | Bedrock Agent SDK 原生 | Anthropic SDK 原生 |
落地建议
-
先用托管方案验证。AgentCore Code Interpreter 几行配置就能跑通 PTC 的完整链路，先确认模型在你的场景下确实能稳定生成可执行代码，再决定是否需要自建沙箱。
-
自建沙箱时先锁网络。ECS Task 放到私有子网，Security Group 只允许出站到 Bedrock API 端点，不允许 SSH 入站。镜像里不装
curl、wget等网络工具——代码执行环境应该是一个计算孤岛。 -
代理路径要加版本对齐测试。Anthropic Messages API 和 Bedrock Converse API 的字段映射不是一劳永逸的。每次 Bedrock 或 Anthropic SDK 升级，跑一遍集成测试确认翻译层没有遗漏新字段。
-
给沙箱加输出大小限制。不管哪种方案，都要限制 stdout/stderr 的返回体积（比如 100 KB），防止模型生成的代码意外输出海量数据，撑爆回传通道。
PTC 把"模型想执行代码"这件事从概念变成了工程问题。三种方案的取舍本质上是控制权与便利性的交换——选哪条路，取决于你愿意为控制权付出多少运维成本，以及你的团队对哪种 SDK 体验更熟悉。