MCP Client SDK使用指南：Python与TypeScript完整教程

引言

MCP Client SDK是连接AI模型与MCP Server的桥梁。本文将分别介绍 Python SDK（mcp 包）和 TypeScript SDK（@modelcontextprotocol/sdk）的完整使用指南。

Python SDK

安装

# 安装官方MCP Python SDK
pip install mcp

# 验证安装
python -c "import mcp; print(mcp.__version__)"

基础使用

连接到stdio Server

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    """连接到stdio MCP Server"""
    # 配置Server参数
    server_params = StdioServerParameters(
        command="node",
        args=["path/to/mcp-server/dist/index.js"],
        env={
            "NODE_ENV": "production",
        }
    )

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 初始化握手
            await session.initialize()

            # 2. 获取工具列表
            tools_result = await session.list_tools()
            print(f"可用工具 ({len(tools_result.tools)} 个):")
            for tool in tools_result.tools:
                print(f"  - {tool.name}: {tool.description}")

            # 3. 调用工具
            result = await session.call_tool(
                "calculate",
                arguments={
                    "operation": "add",
                    "a": 10,
                    "b": 5
                }
            )
            print(f"计算结果: {result.content[0].text}")

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

连接到SSE（HTTP）Server

from mcp.client.sse import sse_client

async def connect_sse():
    """连接到SSE MCP Server"""
    async with sse_client("http://localhost:3000/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 获取可用资源
            resources = await session.list_resources()
            for resource in resources.resources:
                print(f"资源: {resource.name} ({resource.uri})")

            # 读取资源
            content = await session.read_resource(
                "file:///docs/getting-started.md"
            )
            print(content.contents[0].text)

高级用法

错误处理

from mcp.types import (
    ErrorData,
    MethodNotFoundError,
    InvalidParamsError,
)

async def safe_call_tool(session, name: str, arguments: dict):
    """安全调用工具，包含完善的错误处理"""
    try:
        result = await session.call_tool(name, arguments)
        if result.isError:
            print(f"工具执行返回错误: {result.content[0].text}")
            return None
        return result
    except MethodNotFoundError:
        print(f"工具 '{name}' 不存在")
    except InvalidParamsError as e:
        print(f"参数无效: {e}")
    except TimeoutError:
        print(f"工具 '{name}' 调用超时")
    except Exception as e:
        print(f"未知错误: {e}")
    return None

连接池管理

class MCPConnectionPool:
    """MCP连接池 - 复用连接"""
    def __init__(self, server_params: StdioServerParameters, pool_size: int = 3):
        self.server_params = server_params
        self.pool_size = pool_size
        self._connections = []
        self._lock = asyncio.Lock()

    async def get_session(self):
        """获取一个可用会话"""
        async with self._lock:
            # 清理无效连接
            self._connections = [
                c for c in self._connections
                if not c.get("closed", False)
            ]

            if len(self._connections) < self.pool_size:
                conn = await self._create_connection()
                self._connections.append(conn)
                return conn["session"]

            # 复用现有连接
            conn = self._connections.pop(0)
            self._connections.append(conn)
            return conn["session"]

    async def _create_connection(self):
        read, write = await stdio_client(self.server_params).__aenter__()
        session = ClientSession(read, write)
        await session.initialize()
        return {"session": session, "closed": False}

pool = MCPConnectionPool(server_params)
session = await pool.get_session()

TypeScript/JavaScript SDK

安装

npm install @modelcontextprotocol/sdk

基础使用

Client端实现

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

async function main() {
  // 1. 创建传输层
  const transport = new StdioClientTransport({
    command: "node",
    args: ["dist/index.js"],
  });

  // 2. 创建Client
  const client = new Client(
    {
      name: "my-mcp-client",
      version: "1.0.0",
    },
    {
      capabilities: {
        roots: { listChanged: true },
        sampling: {},
      },
    }
  );

  // 3. 建立连接
  await client.connect(transport);

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

  // 5. 调用工具
  const result = await client.callTool({
    name: "calculate",
    arguments: {
      operation: "add",
      a: 10,
      b: 5,
    },
  });

  console.log("结果:", result.content[0].text);

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

main().catch(console.error);

资源访问

// 列出资源
const resources = await client.listResources();
console.log("可用资源:", resources.resources.map(r => r.name));

// 读取资源
const resource = await client.readResource({
  uri: "file:///docs/readme.md",
});
console.log(resource.contents[0].text);

// 订阅资源变更
await client.subscribeResource({
  uri: "file:///config.json",
});

// 监听变更通知
client.on("notifications/resources/changed", (notification) => {
  console.log(`资源变更: ${notification.params.uri}`);
});

提示模板

// 列出提示模板
const prompts = await client.listPrompts();
console.log("提示模板:", prompts.prompts.map(p => p.name));

// 获取提示内容
const prompt = await client.getPrompt({
  name: "code-review",
  arguments: {
    language: "typescript",
    severity: "high",
  },
});
console.log(prompt.messages);

事件处理

// 工具列表变更通知
client.on("notifications/tools/list_changed", () => {
  console.log("工具列表已变更，重新获取...");
  client.listTools().then(tools => {
    console.log("更新后的工具:", tools.tools.map(t => t.name));
  });
});

// 日志通知
client.on("notifications/message", (notification) => {
  const { level, logger, data } = notification.params;
  console.log(`[${level}] ${logger}:`, data);
});

与AI框架集成

LangChain集成

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

async def create_mcp_agent():
    """创建使用MCP工具的LangChain Agent"""
    server_params = StdioServerParameters(
        command="node",
        args=["path/to/server/dist/index.js"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 加载MCP工具为LangChain工具
            tools = await load_mcp_tools(session)

            # 创建Agent
            agent = create_react_agent(model, tools)

            # 使用Agent
            response = await agent.ainvoke({
                "messages": [
                    {"role": "user", "content": "计算 15 * 23 的结果"}
                ]
            })
            return response

OpenAI集成

import OpenAI from "openai";

// 将MCP工具转换为OpenAI Function Calling格式
function convertToOpenAITools(mcpTools: Tool[]): OpenAI.ChatCompletionTool[] {
  return mcpTools.map(tool => ({
    type: "function",
    function: {
      name: tool.name,
      description: tool.description,
      parameters: tool.inputSchema,
    },
  }));
}

async function chatWithMCP(client: Client, userMessage: string) {
  const { tools: mcpTools } = await client.listTools();
  const openaiTools = convertToOpenAITools(mcpTools);

  const response = await openai.chat.completions.create({
    model: "gpt-4",
    messages: [{ role: "user", content: userMessage }],
    tools: openaiTools,
  });

  // 处理工具调用
  const toolCall = response.choices[0].message.tool_calls?.[0];
  if (toolCall) {
    const result = await client.callTool({
      name: toolCall.function.name,
      arguments: JSON.parse(toolCall.function.arguments),
    });
    return result;
  }
}

SDK方法对照表

最佳实践

总结

Python和TypeScript SDK的使用对比：

下一步学习建议：

• 连接管理
• 错误处理策略
• 客户端最佳实践

本文最后更新于 2024-07-12。