自建 MCP Server：Python/TypeScript SDK 完整实现与调试部署

第三十六章：MCP Client 集成：在 Claude Code、桌面应用中连接 MCP

36.1 MCP Client 概述

上一章我们构建了 MCP Server——提供能力的服务端。本章转向另一侧：MCP Client，即如何在实际应用中连接并使用 MCP Server 的能力。

MCP Client 的职责是：

1. 启动并管理 MCP Server 进程（或连接远程 Server）
2. 执行协议握手，协商能力
3. 将 AI 模型的工具调用请求转发给 Server
4. 将 Server 的响应格式化后返回给 AI 模型
5. 管理 Server 生命周期（重启、超时处理）

主要的 MCP Client 宿主应用包括：

• Claude Desktop：图形界面应用，配置最简单
• Claude Code：CLI 工具，支持项目级和全局配置
• 自定义应用：使用 MCP SDK 直接在代码中集成

36.2 Claude Desktop 集成

36.2.1 配置文件位置

Claude Desktop 使用一个 JSON 配置文件管理所有 MCP Server 连接：

36.2.2 完整配置格式

{
  "mcpServers": {
    "服务器名称": {
      "command": "启动命令",
      "args": ["参数1", "参数2"],
      "env": {
        "ENV_VAR": "值"
      },
      "disabled": false,
      "alwaysAllow": ["tool_name_1", "tool_name_2"]
    }
  }
}

字段说明：

• command：启动 Server 的可执行程序（如 python、node、npx）
• args：传递给命令的参数数组
• env：额外的环境变量（会与系统环境变量合并）
• disabled：true 时禁用该 Server 但保留配置
• alwaysAllow：列表中的工具调用无需用户确认（谨慎使用）

36.2.3 常见 Server 配置示例

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/alice/Documents",
        "/Users/alice/Projects"
      ]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_CONNECTION_STRING": "postgresql://user:pass@localhost:5432/mydb"
      }
    },
    "custom-python-server": {
      "command": "python",
      "args": ["/Users/alice/mcp-servers/my_server.py"],
      "env": {
        "WORKSPACE": "/Users/alice/Projects",
        "API_KEY": "your_api_key"
      }
    },
    "brave-search": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-brave-search"],
      "env": {
        "BRAVE_API_KEY": "your_brave_api_key"
      }
    }
  }
}

36.2.4 配置后的验证步骤

1. 保存配置文件
2. 完全退出并重启 Claude Desktop
3. 在新对话中，点击右下角的工具图标（锤子图标）检查 Server 是否连接成功
4. 如果 Server 列表中出现你配置的服务名，说明连接成功
5. 如果连接失败，查看 Claude Desktop 的日志：
   • macOS：~/Library/Logs/Claude/mcp-server-{name}.log

36.2.5 调试连接问题

# 手动测试 Server 能否启动
python /path/to/server.py

# 测试 npx 能否找到包
npx -y @modelcontextprotocol/server-filesystem /tmp

# 检查环境变量是否正确
echo $GITHUB_PERSONAL_ACCESS_TOKEN

常见问题：

• Server 未出现在列表中：通常是启动命令路径错误或依赖未安装
• 工具调用失败：检查 env 中的 API Key 和连接字符串
• 权限错误：确认 Server 对配置路径有读写权限

36.3 Claude Code 中的 MCP 集成

Claude Code 是 Anthropic 的命令行 AI 编程工具，对 MCP 有更灵活的支持，包括项目级配置和动态添加 Server。

36.3.1 全局 MCP 配置

Claude Code 的全局 MCP 配置与 Claude Desktop 共享同一文件（在支持的平台上），也可以在 ~/.claude.json 中单独配置：

# 查看当前所有 MCP 配置
claude mcp list

# 添加 stdio 类型的 Server（全局）
claude mcp add my-server python /path/to/server.py --scope global

# 添加带参数的 Server
claude mcp add filesystem npx -- -y @modelcontextprotocol/server-filesystem /home/user

# 添加带环境变量的 Server
claude mcp add github npx -- -y @modelcontextprotocol/server-github \
  --env GITHUB_PERSONAL_ACCESS_TOKEN=ghp_xxxxx

# 添加 HTTP 类型的远程 Server
claude mcp add remote-server --transport http https://my-mcp-server.example.com

# 删除 Server
claude mcp remove my-server

# 查看特定 Server 的详情
claude mcp get my-server

36.3.2 项目级 MCP 配置

在项目根目录创建 .claude/mcp.json，这样项目内的所有 Claude Code 会话都会自动加载这些 Server：

{
  "mcpServers": {
    "project-tools": {
      "command": "python",
      "args": ["./scripts/mcp_server.py"],
      "env": {
        "PROJECT_ROOT": "."
      }
    },
    "database": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_CONNECTION_STRING": "${DATABASE_URL}"
      }
    }
  }
}

注意：${DATABASE_URL} 这类语法表示从系统环境变量读取，避免将敏感信息写入版本控制。

36.3.3 在 Claude Code 会话中使用 MCP 工具

连接 MCP Server 后，Claude Code 在会话中可以直接调用工具，无需特殊语法：

# 用户对 Claude Code 说：
"请读取 README.md 的内容，然后根据内容更新 package.json 的 description 字段"

# Claude Code 会自动：
# 1. 调用 MCP Server 的 read_file 工具读取 README.md
# 2. 分析内容
# 3. 调用 write_file 工具更新 package.json

也可以在 Claude Code 中显式要求使用特定工具：

"使用 filesystem Server 的 search_files 工具搜索所有包含 'TODO' 的 Python 文件"

36.3.4 CLAUDE.md 与 MCP 集成

在项目的 CLAUDE.md 文件中记录 MCP Server 的使用方法，让 Claude Code 知道何时该用哪个工具：

# 项目 MCP 配置

## 可用的 MCP 工具

### project-tools Server
- `read_schema`：读取数据库 Schema，在处理数据库相关任务前调用
- `run_migration`：执行数据库迁移，需要先备份
- `deploy_preview`：部署到预览环境

### 使用原则
1. 修改数据库前，先调用 `read_schema` 确认表结构
2. 生成代码后，调用 `run_tests` 确认测试通过
3. 只有在用户明确要求时才调用 `deploy_preview`

36.4 在自定义 Python 应用中集成 MCP

36.4.1 使用 MCP Python SDK 作为 Client

# custom_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import anthropic

async def ai_agent_with_mcp(user_request: str):
    """集成 MCP 工具能力的 AI Agent"""
    
    # 启动 MCP Server
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env={"WORKSPACE": "/home/user/project"}
    )
    
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化 MCP 连接
            await session.initialize()
            
            # 获取可用工具
            tools_response = await session.list_tools()
            
            # 将 MCP 工具转换为 Anthropic API 格式
            anthropic_tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema
                }
                for tool in tools_response.tools
            ]
            
            # 与 Claude 交互
            claude_client = anthropic.Anthropic()
            messages = [{"role": "user", "content": user_request}]
            
            while True:
                response = claude_client.messages.create(
                    model="claude-opus-4-5",
                    max_tokens=4096,
                    tools=anthropic_tools,
                    messages=messages
                )
                
                if response.stop_reason == "end_turn":
                    final_text = next(
                        (b.text for b in response.content if b.type == "text"), ""
                    )
                    return final_text
                
                elif response.stop_reason == "tool_use":
                    # 处理工具调用
                    messages.append({
                        "role": "assistant",
                        "content": response.content
                    })
                    
                    tool_results = []
                    for block in response.content:
                        if block.type == "tool_use":
                            # 通过 MCP Client 调用工具
                            mcp_result = await session.call_tool(
                                block.name,
                                block.input
                            )
                            
                            # 提取文本结果
                            result_text = "\n".join([
                                content.text
                                for content in mcp_result.content
                                if hasattr(content, "text")
                            ])
                            
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result_text,
                                "is_error": mcp_result.isError
                            })
                    
                    messages.append({
                        "role": "user",
                        "content": tool_results
                    })


# 使用示例
async def main():
    result = await ai_agent_with_mcp(
        "分析 src/ 目录下所有 Python 文件的代码质量，给出改进建议"
    )
    print(result)

asyncio.run(main())

36.4.2 连接多个 MCP Server

import asyncio
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from typing import Dict, List

class MultiMCPClient:
    """管理多个 MCP Server 连接的客户端"""
    
    def __init__(self):
        self.sessions: Dict[str, ClientSession] = {}
        self.tools: List[dict] = []
        self._exit_stack = AsyncExitStack()
    
    async def connect(self, name: str, params: StdioServerParameters):
        """连接一个 MCP Server"""
        read, write = await self._exit_stack.enter_async_context(
            stdio_client(params)
        )
        session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await session.initialize()
        self.sessions[name] = session
        
        # 收集工具，添加 Server 前缀避免名称冲突
        tools_response = await session.list_tools()
        for tool in tools_response.tools:
            self.tools.append({
                "name": f"{name}__{tool.name}",  # 命名空间隔离
                "description": f"[{name}] {tool.description}",
                "input_schema": tool.inputSchema,
                "_server": name,
                "_original_name": tool.name
            })
        
        print(f"已连接 Server: {name}，加载 {len(tools_response.tools)} 个工具")
    
    async def call_tool(self, namespaced_name: str, arguments: dict) -> str:
        """调用工具（自动路由到正确的 Server）"""
        # 找到对应工具信息
        tool_info = next(
            (t for t in self.tools if t["name"] == namespaced_name),
            None
        )
        if not tool_info:
            raise ValueError(f"未知工具：{namespaced_name}")
        
        server_name = tool_info["_server"]
        original_name = tool_info["_original_name"]
        session = self.sessions[server_name]
        
        result = await session.call_tool(original_name, arguments)
        return "\n".join([
            c.text for c in result.content if hasattr(c, "text")
        ])
    
    async def get_anthropic_tools(self) -> List[dict]:
        """获取 Anthropic API 格式的工具列表"""
        return [
            {
                "name": t["name"],
                "description": t["description"],
                "input_schema": t["input_schema"]
            }
            for t in self.tools
        ]
    
    async def close(self):
        await self._exit_stack.aclose()


async def main():
    client = MultiMCPClient()
    
    try:
        # 连接多个 Server
        await client.connect("files", StdioServerParameters(
            command="python", args=["file_server.py"]
        ))
        await client.connect("db", StdioServerParameters(
            command="npx", args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"POSTGRES_CONNECTION_STRING": "postgresql://..."}
        ))
        
        print(f"共加载 {len(client.tools)} 个工具")
        
        # 使用多服务工具能力
        result = await client.call_tool("files__read_file", {"path": "schema.sql"})
        print(result)
        
    finally:
        await client.close()

asyncio.run(main())

36.5 在自定义 TypeScript 应用中集成 MCP

36.5.1 TypeScript MCP Client 基础

// client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import Anthropic from "@anthropic-ai/sdk";

async function createMCPAgent(
  serverCommand: string,
  serverArgs: string[]
): Promise<void> {
  // 初始化 MCP Client
  const transport = new StdioClientTransport({
    command: serverCommand,
    args: serverArgs,
  });

  const mcpClient = new Client(
    { name: "my-agent", version: "1.0.0" },
    { capabilities: {} }
  );

  await mcpClient.connect(transport);

  // 获取工具列表
  const toolsResponse = await mcpClient.listTools();
  const anthropicTools = toolsResponse.tools.map((tool) => ({
    name: tool.name,
    description: tool.description ?? "",
    input_schema: tool.inputSchema,
  }));

  console.log(`Loaded ${anthropicTools.length} tools from MCP server`);

  // 使用 Claude 进行对话
  const claude = new Anthropic();
  const messages: Anthropic.MessageParam[] = [
    { role: "user", content: "List all Python files in the workspace" },
  ];

  let keepGoing = true;
  while (keepGoing) {
    const response = await claude.messages.create({
      model: "claude-opus-4-5",
      max_tokens: 4096,
      tools: anthropicTools as Anthropic.Tool[],
      messages,
    });

    if (response.stop_reason === "end_turn") {
      const text = response.content
        .filter((b): b is Anthropic.TextBlock => b.type === "text")
        .map((b) => b.text)
        .join("\n");
      console.log("Final answer:", text);
      keepGoing = false;
    } else if (response.stop_reason === "tool_use") {
      messages.push({ role: "assistant", content: response.content });

      const toolResults: Anthropic.ToolResultBlockParam[] = [];
      for (const block of response.content) {
        if (block.type !== "tool_use") continue;

        const mcpResult = await mcpClient.callTool({
          name: block.name,
          arguments: block.input as Record<string, unknown>,
        });

        const resultText = mcpResult.content
          .filter((c): c is { type: "text"; text: string } => c.type === "text")
          .map((c) => c.text)
          .join("\n");

        toolResults.push({
          type: "tool_result",
          tool_use_id: block.id,
          content: resultText,
          is_error: mcpResult.isError ?? false,
        });
      }

      messages.push({ role: "user", content: toolResults });
    } else {
      keepGoing = false;
    }
  }

  await mcpClient.close();
}

createMCPAgent("python", ["server.py"]).catch(console.error);

36.5.2 连接 HTTP+SSE 远程 Server

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

async function connectRemoteMCPServer(serverUrl: string): Promise<Client> {
  const transport = new SSEClientTransport(new URL(serverUrl));

  const client = new Client(
    { name: "remote-client", version: "1.0.0" },
    { capabilities: { sampling: {} } }
  );

  await client.connect(transport);
  console.log(`Connected to remote MCP server: ${serverUrl}`);

  return client;
}

// 使用示例
const client = await connectRemoteMCPServer("https://api.example.com/mcp");
const tools = await client.listTools();
console.log(`Available tools: ${tools.tools.map((t) => t.name).join(", ")}`);

36.6 MCP Server 的访问控制与权限管理

36.6.1 Claude Desktop 的工具确认机制

Claude Desktop 默认在以下情况下要求用户确认：

• 写入或修改文件
• 执行 shell 命令
• 向外部服务发送请求

可以通过 alwaysAllow 字段绕过特定工具的确认（需谨慎）：

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/safe-dir"],
      "alwaysAllow": ["read_file", "list_directory"]
    }
  }
}

36.6.2 Server 端的权限限制

最佳实践是在 Server 端实现权限控制，而不依赖 Client 确认：

# 在 Server 中限制可访问路径
ALLOWED_PATHS = [
    Path("/home/user/projects"),
    Path("/tmp/workspace")
]

def is_path_allowed(path: Path) -> bool:
    resolved = path.resolve()
    return any(
        resolved == allowed or allowed in resolved.parents
        for allowed in ALLOWED_PATHS
    )

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "write_file":
        target = Path(arguments["path"])
        if not is_path_allowed(target):
            return [types.TextContent(
                type="text",
                text=f"权限拒绝：不允许写入路径 {arguments['path']}"
            )]
        # ... 正常处理

36.7 监控与可观测性

36.7.1 记录 MCP 交互日志

import logging
import json

# 配置 MCP 交互日志
mcp_logger = logging.getLogger("mcp.interactions")

@app.call_tool()
async def call_tool_with_logging(name: str, arguments: dict):
    mcp_logger.info(json.dumps({
        "event": "tool_call",
        "tool": name,
        "arguments": {
            # 过滤敏感信息
            k: v for k, v in arguments.items()
            if k not in ["password", "api_key", "token"]
        }
    }))
    
    result = await _actual_call_tool(name, arguments)
    
    mcp_logger.info(json.dumps({
        "event": "tool_result",
        "tool": name,
        "is_error": getattr(result[0], "isError", False) if result else False,
        "result_length": sum(len(r.text) for r in result if hasattr(r, "text"))
    }))
    
    return result

小结

本章全面介绍了 MCP Client 集成的三个层次：Claude Desktop 的 JSON 配置、Claude Code 的 CLI 管理、以及在自定义应用中通过 SDK 编程集成。

关键要点：

1. Claude Desktop 的配置简单但功能有限；Claude Code 支持项目级配置
2. 多 Server 集成时，工具命名空间化（server__tool）可避免冲突
3. 在 Python/TypeScript 应用中集成时，需要手动处理工具调用循环
4. 权限控制应在 Server 端实现，Client 端的确认机制只是辅助
5. 敏感配置（API Key 等）应通过环境变量传递，不要硬编码在配置文件中

下一章将介绍 MCP 官方 Server 生态，包括文件系统、数据库和浏览器控制等开箱即用的 Server。