Skip to content

3.2 第一个 MCP Client——连接 Server 调用工具

上一节小周写了 MCP Server 并在 Claude Desktop 中跑通了。但如果不想依赖 Claude Desktop 呢? 这一节反过来——自己写 Client 连接 Server。


Client 的职责

1建立连接——通过 stdio 或 HTTP 连接到 Server
2初始化协商——交换能力(capability negotiation)
3发现工具——获取 Server 的工具列表和 Schema
4调用工具——转发请求并获取结果

python
# client.py
import asyncio
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class TodoClient:
    def __init__(self):
        self.session = None
        self.exit_stack = AsyncExitStack()

    async def connect(self, server_script: str):
        """连接到 MCP Server"""
        server_params = StdioServerParameters(
            command="python",
            args=[server_script],
            env=None
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        read, write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await self.session.initialize()

        # 发现工具
        response = await self.session.list_tools()
        print(f"已连接,可用工具:{[t.name for t in response.tools]}")

    async def call(self, tool_name: str, arguments: dict = None):
        """调用 MCP 工具"""
        result = await self.session.call_tool(tool_name, arguments or {})
        for content in result.content:
            if hasattr(content, 'text'):
                print(content.text)
        return result

    async def close(self):
        await self.exit_stack.aclose()

async def main():
    client = TodoClient()
    try:
        await client.connect("server.py")  # 上一节的 Server

        # 添加待办
        await client.call("add-todo", {"text": "学 MCP Client 开发"})
        await client.call("add-todo", {"text": "写综合实战"})

        # 列出待办
        print("\n--- 待办列表 ---")
        await client.call("list-todos")
    finally:
        await client.close()

asyncio.run(main())
typescript
// client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

class TodoClient {
  private client: Client;
  private transport: StdioClientTransport | null = null;

  constructor() {
    this.client = new Client({ name: "todo-client", version: "1.0.0" });
  }

  async connect(serverScript: string) {
    this.transport = new StdioClientTransport({
      command: "npx",
      args: ["tsx", serverScript]
    });
    await this.client.connect(this.transport);

    // 发现工具
    const tools = await this.client.listTools();
    console.log("已连接,可用工具:", tools.tools.map(t => t.name));
  }

  async call(toolName: string, args: Record<string, unknown> = {}) {
    const result = await this.client.callTool({ name: toolName, arguments: args });
    for (const content of result.content as any[]) {
      if (content.type === "text") console.log(content.text);
    }
    return result;
  }

  async close() {
    await this.client.close();
  }
}

async function main() {
  const client = new TodoClient();
  try {
    await client.connect("server.ts");

    // 添加待办
    await client.call("add-todo", { text: "学 MCP Client 开发" });
    await client.call("add-todo", { text: "写综合实战" });

    // 列出待办
    console.log("\n--- 待办列表 ---");
    await client.call("list-todos");
  } finally {
    await client.close();
  }
}

main();

关键 API

PythonTypeScript说明
session.initialize()client.connect(transport)初始化连接
session.list_tools()client.listTools()列出可用工具
session.call_tool(name, args)client.callTool({ name, arguments })调用工具
session.list_resources()client.listResources()列出可用资源
session.read_resource(uri)client.readResource({ uri })读取资源
session.list_prompts()client.listPrompts()列出模板
session.get_prompt(name, args)client.getPrompt({ name, arguments })获取模板

连接远程 Server

上面的例子用 stdio 连接本地 Server。如果 Server 部署在远程,需要用 Streamable HTTP:

python
# 连接远程 Server
from mcp.client.streamable_http import streamable_http_client

async with streamable_http_client("http://localhost:8000/mcp") as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()
typescript
// 连接远程 Server
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const transport = new StreamableHTTPClientTransport(
  new URL("http://localhost:8000/mcp")
);
await client.connect(transport);

连接 LLM

👤 用户输入🧠 Claude API🔧 MCP Server📤 结果回传🧠 Claude API💬 回复用户

Client 的真正价值在于连接 LLM。小周写了一个完整的 MCP Client,把 MCP 工具桥接给 Claude API:

python
# chat_loop.py —— 完整的 MCP Client + Claude API chat loop
import asyncio
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def chat_loop():
    # 1. 连接 MCP Server
    server_params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 2. 发现工具并转换为 Claude API 格式
            mcp_tools = await session.list_tools()
            claude_tools = [{
                "name": t.name,
                "description": t.description,
                "input_schema": t.inputSchema
            } for t in mcp_tools.tools]

            anthropic = Anthropic()
            messages = []

            print("已连接 MCP Server,输入消息开始对话(输入 quit 退出)")

            # 3. Chat loop
            while True:
                user_input = input("\n你:")
                if user_input.strip() == "quit":
                    break
                messages.append({"role": "user", "content": user_input})

                # 调用 Claude API(带工具)
                response = anthropic.messages.create(
                    model="claude-sonnet-4-20250514",
                    max_tokens=1000,
                    messages=messages,
                    tools=claude_tools
                )

                # 4. 处理工具调用
                while response.stop_reason == "tool_use":
                    # 把 assistant 的回复加入历史
                    messages.append({"role": "assistant", "content": response.content})

                    tool_results = []
                    for block in response.content:
                        if block.type == "tool_use":
                            print(f"  [调用工具] {block.name}({block.input})")
                            # 通过 MCP 调用工具
                            result = await session.call_tool(block.name, block.input)
                            result_text = "".join(
                                c.text for c in result.content if hasattr(c, "text")
                            )
                            print(f"  [工具结果] {result_text}")
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result_text
                            })

                    # 把工具结果返回给 Claude 继续对话
                    messages.append({"role": "user", "content": tool_results})
                    response = anthropic.messages.create(
                        model="claude-sonnet-4-20250514",
                        max_tokens=1000,
                        messages=messages,
                        tools=claude_tools
                    )

                # 最终文本回复
                for block in response.content:
                    if hasattr(block, "text"):
                        print(f"\nClaude:{block.text}")
                messages.append({"role": "assistant", "content": response.content})

asyncio.run(chat_loop())

这就是 Claude Desktop 内部做的事——只是它有更完善的错误处理、重试逻辑和用户界面。核心流程完全相同:发现工具 → LLM 决定调用 → 通过 MCP 执行 → 结果回传 LLM。


本节核心要点

  • Client 通过 stdio 或 HTTP 连接 Server,初始化后即可发现和调用工具
  • Python 用 ClientSession + stdio_client,TypeScript 用 Client + StdioClientTransport
  • Client 的核心价值:桥接 MCP 工具和 LLM API
  • 远程 Server 用 streamable_http_client / StreamableHTTPClientTransport

练习:写一个完整的 Client 命令行工具(chat loop),让用户输入自然语言,Client 自动桥接 Claude API 和 MCP Server。


← 上一节:第一个 MCP Server | 目录 | 下一节:调试利器 →