在 Agent SDK 中集成 MCP 服务器——连接外部服务扩展代理能力。
MCP 与 Agent SDK
MCP(Model Context Protocol)允许代理连接到外部服务。在 Agent SDK 中使用 MCP:
- 启动 MCP 服务器进程
- 通过标准输入/输出连接
- 发现可用的远程工具
- 在代理循环中调用它们
连接 MCP 服务器
import subprocess
import json
class MCPClient:
def __init__(self, command, args):
self.process = subprocess.Popen(
[command] + args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True
)
def call_tool(self, name, **kwargs):
"""调用 MCP 服务器上的工具。"""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": name,
"arguments": kwargs
}
}
self.process.stdin.write(json.dumps(request) + "\n")
self.process.stdin.flush()
response = json.loads(self.process.stdout.readline())
return response.get("result")
# 使用:连接 GitHub MCP 服务器
mcp = MCPClient("npx", [
"-y", "@modelcontextprotocol/server-github"
])
# 调用 GitHub API
result = mcp.call_tool(
"list_repos",
organization="my-org"
)
整合到代理循环
def agent_with_mcp(prompt):
messages = [{"role": "user", "content": prompt}]
# 注册 MCP 工具
mcp_tools = mcp.discover_tools()
for _ in range(10):
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
tools=mcp_tools + local_tools,
messages=messages
)
tool_uses = [c for c in response.content
if c.type == "tool_use"]
if not tool_uses:
return response.content[0].text
for tool_use in tool_uses:
if is_mcp_tool(tool_use.name):
result = mcp.call_tool(
tool_use.name, **tool_use.input
)
else:
result = execute_local_tool(
tool_use.name, **tool_use.input
)
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(result)
}]
})
下一步
- MCP 协议集成:MCP 详细说明
- Agent SDK 自定义工具