Claude Code SDK #8：MCP 集成全解——三种接入方式 × 工具命名规范 × Tool Search 懒加载，把 Agent 能力边界推到任意外部服务

为什么 MCP 不是"可选插件"

工具命名规范：最容易忽略的细节

mcp__<服务器名>__<工具名>

三种传输方式

• 文档给你的是命令行命令（比如 npx @modelcontextprotocol/server-github）→ 用 stdio
• 文档给你的是URL→ 用 HTTP 或 SSE
• 你想直接在代码里定义工具函数→ 用 SDK 内嵌服务器

stdio：本地子进程

# Python
from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="列出 anthropics/claude-code 最近 3 个 Issues",
    options=ClaudeAgentOptions(
        mcp_servers={
            "github": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-github"],
                "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]}
            }
        },
        allowed_tools=["mcp__github__list_issues"]
    )
):
    ...

// TypeScript
for await (const message of query({
    prompt: "列出最近 3 个 Issues",
    options: {
        mcpServers: {
            github: {
                command: "npx",
                args: ["-y", "@modelcontextprotocol/server-github"],
                env: { GITHUB_TOKEN: process.env.GITHUB_TOKEN }
            }
        },
        allowedTools: ["mcp__github__list_issues"]
    }
})) { ... }

HTTP / SSE：云端服务

options=ClaudeAgentOptions(
    mcp_servers={
        "remote-api": {
            "type": "sse",    # 或 "http" 用于 Streamable HTTP
            "url": "https://api.example.com/mcp/sse",
            "headers": {
                "Authorization": f"Bearer {os.environ['API_TOKEN']}"
            }
        }
    },
    allowed_tools=["mcp__remote-api__*"]
)

SDK 内嵌服务器（Python 专属）

from claude_agent_sdk import tool, create_sdk_mcp_server

@tool("query_db", "按自然语言查询数据库", {"question": str})
async def query_db(args):
    sql = translate_to_sql(args["question"])
    result = await db.execute(sql)
    return {"content": [{"type": "text", "text": str(result)}]}

my_server = create_sdk_mcp_server(
    name="db-tools",
    tools=[query_db]
)

options = ClaudeAgentOptions(
    mcp_servers={"db": my_server},
    allowed_tools=["mcp__db__query_db"]
)

权限控制：为什么不应该依赖 permissionMode

allowed_tools=["mcp__github__*"]   # 只开放 github 服务器
allowed_tools=["mcp__db__query"]   # 只开放 db 服务器的 query 工具

认证：三种方式

环境变量（stdio）

{
    "mcpServers": {
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {
                "GITHUB_TOKEN": "${GITHUB_TOKEN}"
            }
        }
    }
}

HTTP Header（HTTP/SSE）

mcp_servers={
    "secure-api": {
        "type": "http",
        "url": "https://api.example.com/mcp",
        "headers": {"Authorization": f"Bearer {token}"}
    }
}

OAuth2

access_token = await complete_oauth_flow()
mcp_servers={
    "oauth-api": {
        "type": "http",
        "url": "https://api.example.com/mcp",
        "headers": {"Authorization": f"Bearer {access_token}"}
    }
}

连接状态检测与错误处理

async for message in query(prompt="...", options=options):
    # 检查连接状态
    if hasattr(message, 'subtype') and message.subtype == "init":
        servers = message.data.get("mcp_servers", [])
        failed = [s for s in servers if s.get("status") != "connected"]
        if failed:
            print(f"连接失败: {failed}")
            break

if hasattr(message, 'result'):
        print(message.result)

if (message.type === "system" && message.subtype === "init") {
    const failed = message.mcp_servers.filter(s => s.status !== "connected");
    if (failed.length > 0) console.warn("Failed:", failed);
}

• get_mcp_status()：随时获取所有服务器的连接状态
• reconnect_mcp_server(name)：对失败的服务器触发重连
• toggle_mcp_server(name, enabled)：在会话中动态启停某个服务器

Tool Search：上百工具时的上下文管理

1. Agent 遇到需要外部能力的任务
2. 向工具目录发起语义搜索
3. 加载最相关的 3-5 个工具定义
4. 这些工具在后续轮次中保持可用

ClaudeAgentOptions(
    mcp_servers={ ... },
    env={
        # auto:5 = 工具定义超过上下文窗口 5% 时启动懒加载
        "ENABLE_TOOL_SEARCH": "auto:5"
    }
)

发现可用工具

async for message in query(prompt="...", options=options):
    if hasattr(message, 'subtype') and message.subtype == "init":
        mcp_info = message.data.get("mcp_servers", [])
        for server in mcp_info:
            print(f"服务器: {server['name']}")
            for tool in server.get('tools', []):
                print(f"  - {tool['name']}: {tool.get('description', '')}")

实践建议