首页 / AI工具 / MCP Client 开发实战：从零构建支持多 Server...

MCP Client 开发实战：从零构建支持多 Server 的智能助手客户端

作者：1630.top 发布于：2026-07-09 阅读：约 25 分钟 AI工具AI编程

MCP Client 开发实战：从零构建支持多 Server 的智能助手客户端

手把手教你使用 Python MCP SDK 开发一个支持多 MCP Server 连接的智能助手客户端，涵盖连接管理、工具聚合、命名空间隔离与 LLM 集成，附完整可运行代码。

前言

MCP（Model Context Protocol）自发布以来，生态中涌现了大量 MCP Server——数据库查询、文件操作、API 调用、搜索引擎……但大多数教程只关注"如何写一个 MCP Server"，很少有人讲"如何写一个 MCP Client"。

实际上，Client 才是连接 AI 模型与 MCP Server 的桥梁。一个优秀的 MCP Client 需要解决三个核心问题：同时连接多个 Server、聚合工具并消除命名冲突、将工具调用结果回注给 LLM。2026 年 7 月，MCP 协议即将发布重大规范修订，转向无状态架构，移除初始化握手和协议级会话，这对 Client 开发也带来了新的影响。

本文将从零开始，用 Python 构建一个生产级 MCP Client，支持多 Server 连接、工具命名空间隔离和 LLM 智能调度。

---

一、MCP Client 架构设计

1.1 核心概念

MCP 协议采用 Client-Server 架构：

角色职责示例
MCP Server暴露 Tools、Resources、Prompts数据库查询Server、文件操作Server
MCP Client连接Server、聚合工具、调度调用本文要构建的智能助手
LLM接收工具列表、决策调用哪个工具GPT-4o、Claude
用户提问 → MCP Client → 聚合所有 Server 的工具 → 发送给 LLM
                                                          ↓
用户得到回答 ← MCP Client ← 执行工具调用 ← LLM 返回工具调用请求

Client 维护一个 Server 连接池，每个 Server 暴露一组工具。Client 将所有工具聚合后（加上命名空间前缀避免冲突）发送给 LLM，LLM 决定调用哪个工具，Client 负责路由到正确的 Server 执行。

1.3 2026 规范修订的影响

2026 年 7 月 28 日即将发布的新规范将 MCP 转向无状态架构 $TRAE_REF：

这意味着 Client 开发将更简单——无需维护会话状态，连接即用。本文代码兼容当前规范，同时标注新规范下的适配要点。

---

二、环境准备

pip install mcp anthropic openai python-dotenv
mcp-client/
├── client.py           # MCP Client 核心实现
├── config.json         # Server 配置文件
├── .env                # 环境变量（API密钥）
└── example_servers/    # 示例 MCP Server
    ├── calculator.py
    └── file_search.py
{
  "servers": {
    "calculator": {
      "command": "python3",
      "args": ["example_servers/calculator.py"],
      "description": "数学计算工具"
    },
    "file_search": {
      "command": "python3",
      "args": ["example_servers/file_search.py"],
      "description": "文件搜索工具"
    }
  },
  "llm": {
    "provider": "anthropic",
    "model": "claude-sonnet-4-20250514"
  }
}

---

三、完整 Client 代码实现

#!/usr/bin/env python3
"""mcp-client/client.py - 多 Server MCP 智能助手客户端"""
import asyncio
import json
import os
from contextlib import AsyncExitStack
from typing import Any, Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()


class MCPClient:
    """支持多 Server 连接的 MCP 客户端"""

    def __init__(self, config_path: str = "config.json"):
        self.config = self._load_config(config_path)
        self.exit_stack = AsyncExitStack()
        self.sessions: dict[str, ClientSession] = {}  # server_name -> session
        self.all_tools: list[dict] = []  # 聚合的所有工具
        self.tool_server_map: dict[str, str] = {}  # tool_name -> server_name
        self.anthropic = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

    def _load_config(self, path: str) -> dict:
        with open(path, "r") as f:
            return json.load(f)

    async def connect_server(self, name: str, command: str, args: list[str], env: dict = None):
        """连接单个 MCP Server"""
        server_params = StdioServerParameters(
            command=command,
            args=args,
            env={**os.environ, **(env or {})}
        )

        # 建立 stdio 连接
        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        read_stream, write_stream = stdio_transport

        # 创建会话
        session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )

        # 初始化会话（新规范将移除此步骤）
        await session.initialize()

        # 获取工具列表
        tools_response = await session.list_tools()
        server_tools = tools_response.tools

        # 为每个工具添加命名空间前缀，避免命名冲突
        for tool in server_tools:
            namespaced_name = f"{name}__{tool.name}"
            self.all_tools.append({
                "name": namespaced_name,
                "description": f"[{name}] {tool.description}",
                "input_schema": tool.inputSchema,
            })
            self.tool_server_map[namespaced_name] = name

        self.sessions[name] = session
        print(f"  [Connected] {name}: {len(server_tools)} tools")

    async def connect_all(self):
        """连接配置文件中的所有 Server"""
        print("Connecting to MCP servers...")
        for name, cfg in self.config["servers"].items():
            try:
                await self.connect_server(
                    name=name,
                    command=cfg["command"],
                    args=cfg.get("args", []),
                    env=cfg.get("env")
                )
            except Exception as e:
                print(f"  [Failed] {name}: {e}")
        print(f"Total tools available: {len(self.all_tools)}")

    async def call_tool(self, tool_name: str, arguments: dict) -> str:
        """调用工具（自动路由到正确的 Server）"""
        # 解析命名空间
        if "__" in tool_name:
            server_name, original_name = tool_name.split("__", 1)
        else:
            # 尝试直接匹配
            server_name = self.tool_server_map.get(tool_name, "")
            original_name = tool_name

        if server_name not in self.sessions:
            return f"Error: Server '{server_name}' not connected"

        session = self.sessions[server_name]
        try:
            result = await session.call_tool(original_name, arguments)
            # 提取文本内容
            if result.content:
                texts = []
                for item in result.content:
                    if hasattr(item, "text"):
                        texts.append(item.text)
                    else:
                        texts.append(str(item))
                return "\n".join(texts)
            return "Tool returned no content"
        except Exception as e:
            return f"Tool error: {e}"

    def _build_anthropic_tools(self) -> list[dict]:
        """将 MCP 工具转换为 Anthropic API 格式"""
        return [
            {
                "name": tool["name"],
                "description": tool["description"],
                "input_schema": tool["input_schema"],
            }
            for tool in self.all_tools
        ]

    async def chat(self, user_message: str, max_turns: int = 10) -> str:
        """与 LLM 进行多轮工具调用对话"""
        messages = [{"role": "user", "content": user_message}]
        system_prompt = (
            "你是一个智能助手，可以通过 MCP 工具完成任务。"
            "工具名格式为 server__tool，请根据描述选择合适的工具。"
            "如果不需要工具，直接回答即可。"
        )

        for turn in range(max_turns):
            response = self.anthropic.messages.create(
                model=self.config["llm"]["model"],
                system=system_prompt,
                messages=messages,
                tools=self._build_anthropic_tools(),
                max_tokens=4096,
            )

            # 如果模型没有请求工具调用，返回最终回答
            if response.stop_reason == "end_turn":
                return response.content[0].text

            # 处理工具调用
            assistant_content = []
            tool_results = []

            for block in response.content:
                assistant_content.append(block)

                if block.type == "tool_use":
                    tool_name = block.name
                    tool_args = block.input

                    print(f"  [Tool Call] {tool_name}({json.dumps(tool_args, ensure_ascii=False)})")

                    # 执行工具调用
                    result = await self.call_tool(tool_name, tool_args)
                    print(f"  [Tool Result] {result[:200]}...")

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

            # 将助手回复和工具结果加入消息历史
            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})

        return "达到最大轮次限制，未能完成任务。"

    async def cleanup(self):
        """清理所有连接"""
        await self.exit_stack.aclose()


async def main():
    client = MCPClient("config.json")

    try:
        await client.connect_all()

        # 交互式对话
        print("\n=== MCP 智能助手 ===")
        print("输入问题开始对话，输入 'quit' 退出\n")

        while True:
            user_input = input("You> ").strip()
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if not user_input:
                continue

            print("Assistant> ", end="", flush=True)
            response = await client.chat(user_input)
            print(response)
            print()

    finally:
        await client.cleanup()
        print("\nDisconnected from all servers.")


if __name__ == "__main__":
    asyncio.run(main())
#!/usr/bin/env python3
"""example_servers/calculator.py - 计算器 MCP Server"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json

server = Server("calculator")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="add",
            description="两个数字相加",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "第一个数"},
                    "b": {"type": "number", "description": "第二个数"},
                },
                "required": ["a", "b"],
            },
        ),
        Tool(
            name="multiply",
            description="两个数字相乘",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "第一个数"},
                    "b": {"type": "number", "description": "第二个数"},
                },
                "required": ["a", "b"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "add":
        result = arguments["a"] + arguments["b"]
    elif name == "multiply":
        result = arguments["a"] * arguments["b"]
    else:
        result = f"Unknown tool: {name}"
    return [TextContent(type="text", text=str(result))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
#!/usr/bin/env python3
"""example_servers/file_search.py - 文件搜索 MCP Server"""
import os
import fnmatch
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("file_search")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_files",
            description="在指定目录中搜索匹配模式的文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "directory": {"type": "string", "description": "搜索目录路径"},
                    "pattern": {"type": "string", "description": "文件名匹配模式，如 *.py"},
                },
                "required": ["directory", "pattern"],
            },
        ),
        Tool(
            name="read_file",
            description="读取文件内容（限制前100行）",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                },
                "required": ["path"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_files":
        directory = arguments["directory"]
        pattern = arguments["pattern"]
        results = []
        for root, dirs, files in os.walk(directory):
            for fname in fnmatch.filter(files, pattern):
                results.append(os.path.join(root, fname))
            if len(results) >= 50:
                break
        text = f"Found {len(results)} files:\n" + "\n".join(results[:50])
        return [TextContent(type="text", text=text)]

    elif name == "read_file":
        path = arguments["path"]
        try:
            with open(path, "r") as f:
                lines = [f.readline() for _ in range(100)]
            return [TextContent(type="text", text="".join(lines))]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {e}")]

    return [TextContent(type="text", text=f"Unknown tool: {name}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

---

四、运行与测试

# 设置 API 密钥
export ANTHROPIC_API_KEY="sk-ant-xxx"

# 启动
python3 client.py
=== MCP 智能助手 ===
输入问题开始对话，输入 'quit' 退出

You> 计算 123 乘以 456，再加上 789
  [Tool Call] calculator__multiply({"a": 123, "b": 456})
  [Tool Result] 56088...
  [Tool Call] calculator__add({"a": 56088, "b": 789})
  [Tool Result] 56877...
Assistant> 123 × 456 = 56088，再加上 789 等于 56877。

You> 搜索当前目录下所有 Python 文件
  [Tool Call] file_search__search_files({"directory": ".", "pattern": "*.py"})
  [Tool Result] Found 3 files:
  ./client.py
  ./example_servers/calculator.py
  ./example_servers/file_search.py...
Assistant> 在当前目录下找到 3 个 Python 文件：client.py、calculator.py 和 file_search.py。

---

五、高级特性

5.1 工具过滤与按需加载

class SmartMCPClient(MCPClient):
    """支持按需加载工具的智能客户端"""

    async def get_relevant_tools(self, user_message: str) -> list[dict]:
        """根据用户问题筛选相关工具"""
        # 方法1：基于关键词匹配
        relevant = []
        for tool in self.all_tools:
            desc = tool["description"].lower()
            if any(kw in user_message.lower() for kw in desc.split()):
                relevant.append(tool)

        # 方法2：用小模型做工具路由（更准确但增加延迟）
        # router_response = self.anthropic.messages.create(
        #     model="claude-3-5-haiku-20241022",
        #     messages=[{"role": "user", "content": f"问题: {user_message}\n选择相关工具: {json.dumps(self.all_tools)}"}],
        #     max_tokens=200,
        # )

        # 如果没有匹配到，返回所有工具
        return relevant if relevant else self.all_tools

    async def chat(self, user_message: str, max_turns: int = 10) -> str:
        # 动态筛选工具
        relevant_tools = await self.get_relevant_tools(user_message)
        print(f"  [Tool Filter] {len(relevant_tools)}/{len(self.all_tools)} tools selected")

        # 使用筛选后的工具进行对话
        # ... (其余逻辑与基类相同，但使用 relevant_tools 替代 self.all_tools)

5.2 支持 SSE 和 HTTP 传输

from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamablehttp_client

async def connect_sse_server(self, name: str, url: str):
    """通过 SSE 连接远程 MCP Server"""
    sse_transport = await self.exit_stack.enter_async_context(
        sse_client(url)
    )
    read_stream, write_stream = sse_transport
    session = await self.exit_stack.enter_async_context(
        ClientSession(read_stream, write_stream)
    )
    await session.initialize()
    # ... 注册工具逻辑同上

async def connect_http_server(self, name: str, url: str):
    """通过 Streamable HTTP 连接远程 MCP Server"""
    http_transport = await self.exit_stack.enter_async_context(
        streamablehttp_client(url)
    )
    read_stream, write_stream, _ = http_transport
    session = await self.exit_stack.enter_async_context(
        ClientSession(read_stream, write_stream)
    )
    await session.initialize()
    # ... 注册工具逻辑同上
async def connect_with_retry(self, name: str, max_retries: int = 3, **kwargs):
    """带重试的连接"""
    for attempt in range(max_retries):
        try:
            await self.connect_server(name, **kwargs)
            return
        except Exception as e:
            print(f"  [Retry {attempt+1}/{max_retries}] {name}: {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # 指数退避
    print(f"  [Failed] {name}: max retries exceeded")

async def health_check(self):
    """定期检查所有 Server 连接健康状态"""
    while True:
        await asyncio.sleep(60)  # 每60秒检查一次
        for name, session in list(self.sessions.items()):
            try:
                await session.list_tools()  # 简单的心跳检测
            except Exception:
                print(f"  [Health Check] {name} disconnected, reconnecting...")
                # 重新连接逻辑

---

六、实操部署步骤

mkdir mcp-client && cd mcp-client
python3 -m venv venv
source venv/bin/activate
pip install mcp anthropic python-dotenv
cat > .env << 'EOF'
ANTHROPIC_API_KEY=sk-ant-your-key-here
EOF
python3 client.py
docker build -t mcp-client .
docker run -it --env-file .env mcp-client

---

七、常见问题 FAQ

Q1: 多个 Server 有同名工具怎么办？

Client 通过命名空间前缀解决：工具名格式为 server__tool。例如 calculator__add 和 math__add 不会冲突。LLM 在调用时使用带前缀的完整名称。

Q2: 工具太多导致 LLM 上下文溢出怎么办？

实现按需加载策略：1) 用关键词匹配筛选相关工具；2) 用小模型（如 Haiku）做工具路由预筛选；3) 分组加载，先加载常用工具，按需追加。

Q3: 2026 新规范对现有 Client 有什么影响？

新规范移除了 initialize 握手和协议级会话，转为无状态架构 $TRAE_REF。现有 Client 代码在新 SDK 中只需移除 await session.initialize() 调用即可。Server 端可自由水平扩展，无需维护会话状态。

Q4: 如何支持 OpenAI 格式的 LLM？

将 _build_anthropic_tools() 的输出转换为 OpenAI Function Calling 格式即可。OpenAI 使用 function 对象包裹工具定义，调用时通过 tool_calls 字段返回。核心逻辑（工具聚合、命名空间、路由执行）完全相同。

Q5: MCP Server 崩溃了如何处理？

实现健康检查机制（如上文 health_check 方法），定期检测连接状态。检测到断连后自动重连，使用指数退避策略避免雪崩。对于关键 Server，可以启动多个实例做故障转移。

---

八、总结

本文从零构建了一个支持多 Server 连接的 MCP Client，核心设计要点：

MCP Client 是连接 AI 与工具生态的关键组件。随着 2026 年新规范的发布，Client 开发将更加简洁，Server 扩展性也将大幅提升。掌握 Client 开发，你就能构建属于自己的 AI 工具调度中枢。