首页 教程 常见问题

Agent SDK MCP 集成

在 Agent SDK 中集成 MCP 服务器——连接外部服务扩展代理能力。

MCP 与 Agent SDK

MCP(Model Context Protocol)允许代理连接到外部服务。在 Agent SDK 中使用 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)
                }]
            })

下一步