MCP 客户端规范

本节包含模型上下文协议 (MCP) 客户端实现的详细规范。

概述

MCP 客户端是连接到 MCP 服务器并使用其功能的应用程序。客户端负责：

• 建立与服务器的连接
• 发送请求并处理响应
• 管理会话状态
• 处理错误和重连

客户端类型

1. AI 助手客户端

• Claude Desktop: Anthropic 的官方桌面应用
• ChatGPT: OpenAI 的聊天界面
• 其他 AI 助手: 集成 MCP 的各种 AI 应用

2. 开发工具客户端

• IDE 插件: VS Code、IntelliJ 等
• 命令行工具: MCP CLI 等
• 调试工具: MCP Inspector 等

3. 自定义客户端

• Web 应用: 基于浏览器的客户端
• 移动应用: iOS/Android 客户端
• 桌面应用: 原生桌面客户端

核心组件

连接管理

interface MCPClient {
    // 连接到服务器
    connect(transport: Transport): Promise<void>;
    
    // 断开连接
    disconnect(): Promise<void>;
    
    // 检查连接状态
    isConnected(): boolean;
    
    // 重新连接
    reconnect(): Promise<void>;
}

请求处理

interface RequestHandler {
    // 发送请求
    sendRequest<T>(method: string, params?: any): Promise<T>;
    
    // 发送通知
    sendNotification(method: string, params?: any): Promise<void>;
    
    // 处理服务器通知
    onNotification(method: string, handler: (params: any) => void): void;
}

会话管理

interface SessionManager {
    // 初始化会话
    initialize(): Promise<InitializeResult>;
    
    // 获取服务器信息
    getServerInfo(): ServerInfo;
    
    // 管理会话状态
    getSessionState(): SessionState;
}

实现要求

必需功能

• [x] JSON-RPC 2.0 协议支持
• [x] 初始化握手
• [x] 错误处理
• [x] 连接管理

推荐功能

• [x] 自动重连
• [x] 请求超时
• [x] 批量请求
• [x] 流式响应

可选功能

• [ ] 缓存机制
• [ ] 离线模式
• [ ] 多服务器支持
• [ ] 插件系统

规范文档

基础规范

• 连接管理 - 连接建立和管理
• 初始化流程 - 客户端初始化
• 请求处理 - 请求和响应处理

高级功能

• 错误处理 - 错误处理策略
• 重连机制 - 自动重连实现
• 性能优化 - 性能优化指南

安全规范

• 身份验证 - 客户端身份验证
• 授权管理 - 权限管理
• 安全传输 - 安全通信

实现示例

Python 客户端

import asyncio
from mcp import Client, StdioTransport

async def main():
    # 创建传输层
    transport = StdioTransport("python", ["-m", "my_server"])
    
    # 创建客户端
    client = Client("my-client")
    
    try:
        # 连接到服务器
        await client.connect(transport)
        
        # 初始化
        result = await client.initialize()
        print(f"连接到服务器: {result.serverInfo.name}")
        
        # 列出可用工具
        tools = await client.list_tools()
        print(f"可用工具: {[tool.name for tool in tools.tools]}")
        
        # 调用工具
        if tools.tools:
            result = await client.call_tool(
                tools.tools[0].name,
                {"param": "value"}
            )
            print(f"工具结果: {result}")
            
    finally:
        await client.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

TypeScript 客户端

import { Client, StdioTransport } from "@modelcontextprotocol/sdk";

async function main() {
    // 创建传输层
    const transport = new StdioTransport({
        command: "python",
        args: ["-m", "my_server"]
    });
    
    // 创建客户端
    const client = new Client({
        name: "my-client",
        version: "1.0.0"
    });
    
    try {
        // 连接到服务器
        await client.connect(transport);
        
        // 初始化
        const result = await client.initialize();
        console.log(`连接到服务器: ${result.serverInfo.name}`);
        
        // 列出可用资源
        const resources = await client.listResources();
        console.log(`可用资源: ${resources.resources.map(r => r.name)}`);
        
        // 读取资源
        if (resources.resources.length > 0) {
            const content = await client.readResource(
                resources.resources[0].uri
            );
            console.log(`资源内容: ${content}`);
        }
        
    } finally {
        await client.disconnect();
    }
}

main().catch(console.error);

测试指南

单元测试

import pytest
from unittest.mock import Mock, AsyncMock
from mcp import Client

@pytest.mark.asyncio
async def test_client_initialization():
    # 模拟传输层
    transport = Mock()
    transport.connect = AsyncMock()
    transport.send = AsyncMock()
    transport.receive = AsyncMock(return_value={
        "jsonrpc": "2.0",
        "id": 1,
        "result": {
            "protocolVersion": "2024-11-05",
            "serverInfo": {"name": "test-server", "version": "1.0.0"},
            "capabilities": {}
        }
    })
    
    # 测试客户端初始化
    client = Client("test-client")
    await client.connect(transport)
    
    result = await client.initialize()
    assert result.serverInfo.name == "test-server"
    
    await client.disconnect()

集成测试

import asyncio
import subprocess
from mcp import Client, StdioTransport

async def test_integration():
    # 启动测试服务器
    server_process = subprocess.Popen([
        "python", "-m", "test_server"
    ])
    
    try:
        # 等待服务器启动
        await asyncio.sleep(1)
        
        # 连接到服务器
        transport = StdioTransport("python", ["-m", "test_server"])
        client = Client("test-client")
        
        await client.connect(transport)
        await client.initialize()
        
        # 测试功能
        tools = await client.list_tools()
        assert len(tools.tools) > 0
        
        await client.disconnect()
        
    finally:
        server_process.terminate()
        server_process.wait()

最佳实践

连接管理

1. 优雅关闭: 总是正确关闭连接
2. 错误恢复: 实现自动重连机制
3. 超时处理: 设置合理的超时时间
4. 资源清理: 及时释放资源

性能优化

1. 连接复用: 复用现有连接
2. 批量请求: 合并多个请求
3. 缓存策略: 缓存常用数据
4. 异步处理: 使用异步 I/O

错误处理

1. 分类处理: 区分不同类型的错误
2. 重试机制: 实现指数退避重试
3. 用户反馈: 提供清晰的错误信息
4. 日志记录: 记录详细的错误日志

相关资源

• 服务器规范
• 协议规范
• 传输层
• SDK 文档