第四章：客户端集成

MCP 客户端概述

MCP 客户端负责与服务器建立连接、管理通信，并将服务能力暴露给宿主应用。

客户端职责

┌─────────────────────────────────────────────────────┐
│                   宿主应用                           │
│  (Claude Desktop, IDE, 自定义应用)                   │
└─────────────────────────────────────────────────────┘
                        ↓
┌─────────────────────────────────────────────────────┐
│                   MCP Client                         │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐   │
│  │ 连接管理    │ │ 能力协商    │ │ 消息路由    │   │
│  └─────────────┘ └─────────────┘ └─────────────┘   │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐   │
│  │ 资源访问    │ │ 工具调用    │ │ 提示词获取  │   │
│  └─────────────┘ └─────────────┘ └─────────────┘   │
└─────────────────────────────────────────────────────┘
                        ↓
                   MCP Server

Python 客户端

基础连接

from mcp import Client
from mcp.client.stdio import stdio_client

async def connect_to_server():
    # 启动并连接到服务器
    async with stdio_client(
        "python", ["my_server.py"]
    ) as (read, write):
        async with Client(read, write) as client:
            # 初始化连接
            result = await client.initialize()
            print(f"已连接到: {result.serverInfo.name}")

            # 使用客户端...
            tools = await client.list_tools()
            print(f"可用工具: {[t.name for t in tools]}")

完整客户端示例

import asyncio
from mcp import Client
from mcp.client.stdio import stdio_client

class MCPClientWrapper:
    def __init__(self, command: str, args: list):
        self.command = command
        self.args = args
        self.client = None

    async def connect(self):
        """建立连接"""
        self.read, self.write = await stdio_client(
            self.command, self.args
        ).__aenter__()
        self.client = Client(self.read, self.write)
        await self.client.__aenter__()
        self.init_result = await self.client.initialize()

    async def disconnect(self):
        """断开连接"""
        if self.client:
            await self.client.__aexit__(None, None, None)

    async def list_tools(self):
        """列出工具"""
        return await self.client.list_tools()

    async def call_tool(self, name: str, arguments: dict):
        """调用工具"""
        return await self.client.call_tool(name, arguments)

    async def list_resources(self):
        """列出资源"""
        return await self.client.list_resources()

    async def read_resource(self, uri: str):
        """读取资源"""
        return await self.client.read_resource(uri)

# 使用示例
async def main():
    client = MCPClientWrapper("python", ["my_server.py"])
    await client.connect()

    print(f"服务器: {client.init_result.serverInfo.name}")

    # 列出工具
    tools = await client.list_tools()
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")

    # 调用工具
    result = await client.call_tool("echo", {"message": "Hello"})
    print(f"结果: {result.content[0].text}")

    await client.disconnect()

asyncio.run(main())

TypeScript 客户端

安装

npm install @modelcontextprotocol/sdk

基础使用

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function main() {
  // 创建传输层
  const transport = new StdioClientTransport({
    command: "python",
    args: ["my_server.py"],
  });

  // 创建客户端
  const client = new Client(
    {
      name: "my-client",
      version: "1.0.0",
    },
    {
      capabilities: {},
    }
  );

  // 连接
  await client.connect(transport);

  // 列出工具
  const tools = await client.listTools();
  console.log("可用工具:", tools.tools.map((t) => t.name));

  // 调用工具
  const result = await client.callTool({
    name: "echo",
    arguments: { message: "Hello from TypeScript" },
  });
  console.log("结果:", result.content);

  // 关闭连接
  await client.close();
}

main();

完整客户端类

import {
  Client,
  ListToolsResult,
  CallToolResult,
  ListResourcesResult,
  ReadResourceResult,
} from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

export class MCPClient {
  private client: Client;
  private transport: StdioClientTransport;

  constructor(
    private command: string,
    private args: string[]
  ) {}

  async connect(): Promise<void> {
    this.transport = new StdioClientTransport({
      command: this.command,
      args: this.args,
    });

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

    await this.client.connect(this.transport);
  }

  async disconnect(): Promise<void> {
    await this.client.close();
  }

  async listTools(): Promise<ListToolsResult> {
    return this.client.listTools();
  }

  async callTool(
    name: string,
    args: Record<string, unknown>
  ): Promise<CallToolResult> {
    return this.client.callTool({ name, arguments: args });
  }

  async listResources(): Promise<ListResourcesResult> {
    return this.client.listResources();
  }

  async readResource(uri: string): Promise<ReadResourceResult> {
    return this.client.readResource({ uri });
  }

  async listPrompts() {
    return this.client.listPrompts();
  }

  async getPrompt(name: string, args?: Record<string, string>) {
    return this.client.getPrompt({ name, arguments: args });
  }
}

// 使用示例
async function example() {
  const mcp = new MCPClient("python", ["my_server.py"]);
  await mcp.connect();

  // 获取并显示工具
  const { tools } = await mcp.listTools();
  console.log(`发现 ${tools.length} 个工具`);

  // 调用工具
  const result = await mcp.callTool("search", { query: "MCP" });
  console.log(result.content);

  await mcp.disconnect();
}

连接多个服务器

多服务器管理

class MultiServerClient:
    def __init__(self):
        self.servers = {}

    async def add_server(self, name: str, command: str, args: list):
        """添加服务器"""
        client = MCPClientWrapper(command, args)
        await client.connect()
        self.servers[name] = client

    async def remove_server(self, name: str):
        """移除服务器"""
        if name in self.servers:
            await self.servers[name].disconnect()
            del self.servers[name]

    async def list_all_tools(self):
        """列出所有服务器的工具"""
        all_tools = {}
        for name, client in self.servers.items():
            tools = await client.list_tools()
            all_tools[name] = tools
        return all_tools

    async def call_tool(self, server_name: str, tool_name: str, args: dict):
        """调用指定服务器的工具"""
        if server_name not in self.servers:
            raise ValueError(f"服务器 {server_name} 未连接")
        return await self.servers[server_name].call_tool(tool_name, args)

# 使用示例
async def main():
    manager = MultiServerClient()

    # 连接多个服务器
    await manager.add_server("filesystem", "mcp-filesystem", ["/home/user"])
    await manager.add_server("github", "mcp-github", [])
    await manager.add_server("database", "python", ["db_server.py"])

    # 列出所有工具
    all_tools = await manager.list_all_tools()
    for server, tools in all_tools.items():
        print(f"{server}: {[t.name for t in tools]}")

    # 调用特定服务器的工具
    result = await manager.call_tool("filesystem", "read_file", {"path": "README.md"})

处理通知

接收服务器通知

from mcp.types import ResourceUpdatedNotification

async def handle_notification(notification):
    """处理服务器通知"""
    if isinstance(notification, ResourceUpdatedNotification):
        print(f"资源更新: {notification.uri}")
        # 刷新缓存或通知用户

async def run_client_with_notifications():
    async with stdio_client("python", ["server.py"]) as (read, write):
        async with Client(read, write) as client:
            # 设置通知处理器
            client.on_notification(handle_notification)

            await client.initialize()

            # 主循环
            while True:
                await asyncio.sleep(1)

TypeScript 通知处理

client.setNotificationHandler(
  ResourceUpdatedNotificationSchema,
  async (notification) => {
    console.log(`资源更新: ${notification.params.uri}`);
    // 处理更新
  }
);

client.setNotificationHandler(
  ToolListChangedNotificationSchema,
  async () => {
    console.log("工具列表已变更");
    const { tools } = await client.listTools();
    // 更新工具列表
  }
);

错误处理

连接错误

async def robust_connect(command: str, args: list, max_retries: int = 3):
    """带重试的连接"""
    for attempt in range(max_retries):
        try:
            client = MCPClientWrapper(command, args)
            await client.connect()
            return client
        except Exception as e:
            print(f"连接失败 (尝试 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # 指数退避
    raise ConnectionError("无法连接到服务器")

工具调用错误

async def safe_call_tool(client: MCPClientWrapper, name: str, args: dict):
    """安全的工具调用"""
    try:
        result = await client.call_tool(name, args)
        if result.isError:
            print(f"工具执行错误: {result.content[0].text}")
            return None
        return result
    except Exception as e:
        print(f"调用失败: {e}")
        return None

与 LLM 集成

OpenAI 集成

from openai import OpenAI

async def use_mcp_with_openai(mcp_client: MCPClientWrapper, user_message: str):
    """将 MCP 工具与 OpenAI 结合"""
    openai = OpenAI()

    # 获取 MCP 工具
    mcp_tools = await mcp_client.list_tools()

    # 转换为 OpenAI 工具格式
    openai_tools = [
        {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        }
        for tool in mcp_tools
    ]

    # 调用 OpenAI
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}],
        tools=openai_tools
    )

    # 处理工具调用
    if response.choices[0].message.tool_calls:
        for tool_call in response.choices[0].message.tool_calls:
            result = await mcp_client.call_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            print(f"工具结果: {result.content[0].text}")

Anthropic Claude 集成

from anthropic import Anthropic

async def use_mcp_with_claude(mcp_client: MCPClientWrapper, user_message: str):
    """将 MCP 工具与 Claude 结合"""
    anthropic = Anthropic()

    # 获取 MCP 工具
    mcp_tools = await mcp_client.list_tools()

    # 转换为 Claude 工具格式
    claude_tools = [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema
        }
        for tool in mcp_tools
    ]

    # 调用 Claude
    response = anthropic.messages.create(
        model="claude-3-opus-20240229",
        max_tokens=4096,
        messages=[{"role": "user", "content": user_message}],
        tools=claude_tools
    )

    # 处理工具调用
    for block in response.content:
        if block.type == "tool_use":
            result = await mcp_client.call_tool(
                block.name,
                block.input
            )
            print(f"工具结果: {result.content[0].text}")

小结

MCP 客户端集成要点：

• Python 和 TypeScript 都有官方 SDK
• 支持连接多个服务器
• 正确处理通知和错误
• 可与各种 LLM 集成

下一章我们将学习资源与工具的详细用法。