跳至主要內容

MCP 客户端开发与集成

HSXISAWD大约 8 分钟AIMCPClient 开发LLM 集成Claude API

MCP 客户端开发与集成

客户端开发概述

MCP Client 是连接 MCP Server 和 LLM 的桥梁。它的核心职责是：

Client 的关键工作流程：

  1. 连接 Server — 建立传输通道，完成初始化和能力协商
  2. 发现工具 — 调用 tools/list 获取 Server 提供的工具
  3. 格式转换 — 将 MCP Tool 定义转换为 LLM 的 Function Calling 格式
  4. 调用工具 — 将 LLM 的工具调用请求转发给 Server
  5. 回传结果 — 将 Server 的执行结果返回给 LLM

Python 客户端开发

基础客户端结构

import asyncio
import sys
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class MCPClient:
    """MCP 客户端 — 连接 Server 并与 LLM 交互"""

    def __init__(self):
        self.session: ClientSession | None = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script: str):
        """连接到 MCP Server"""
        # 根据文件扩展名判断启动命令
        command = "python" if server_script.endswith(".py") else "node"

        server_params = StdioServerParameters(
            command=command,
            args=[server_script],
            env=None,  # 继承当前环境变量
        )

        # 启动 Server 子进程并建立连接
        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        read_stream, write_stream = stdio_transport

        # 创建 Session 并初始化
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        await self.session.initialize()

        # 列出可用工具
        response = await self.session.list_tools()
        print(f"已连接到 Server，发现 {len(response.tools)} 个工具：")
        for tool in response.tools:
            print(f"  - {tool.name}: {tool.description}")

    async def call_tool(self, name: str, arguments: dict) -> str:
        """调用 MCP 工具"""
        if not self.session:
            return "错误：未连接到 Server"

        result = await self.session.call_tool(name, arguments)
        # 提取文本内容
        texts = []
        for content in result.content:
            if hasattr(content, "text"):
                texts.append(content.text)
        return "\n".join(texts)

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("server.py")

        # 调用工具示例
        result = await client.call_tool("read_file", {"filepath": "test.txt"})
        print(f"结果: {result}")
    finally:
        await client.cleanup()


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

连接生命周期管理

[!warning] 资源管理
使用 AsyncExitStack 管理 MCP 连接的生命周期非常重要。它确保在发生异常时也能正确关闭连接和子进程，避免资源泄露。

集成 LLM：完整的 Agent 客户端

使用 Claude API 集成

以下是一个完整的 MCP Client，将 MCP 工具桥接到 Claude API：

import asyncio
import json
import sys
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic


class MCPAgent:
    """MCP + Claude Agent — 完整的 AI Agent 客户端"""

    def __init__(self):
        self.session: ClientSession | None = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
        self.model = "claude-sonnet-4-20250514"

    async def connect_to_server(self, server_script: str):
        """连接到 MCP Server"""
        command = "python" if server_script.endswith(".py") else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script],
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        read_stream, write_stream = stdio_transport

        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        await self.session.initialize()

    def _get_available_tools(self) -> list[dict]:
        """将 MCP 工具转换为 Claude Function Calling 格式"""
        if not self.session:
            return []

        # 注意：这里需要先调用 session.list_tools() 获取工具列表
        # 实际使用中应该缓存这个结果
        return self._tools_cache

    async def _cache_tools(self):
        """缓存工具列表"""
        response = await self.session.list_tools()
        self._tools_cache = [
            {
                "name": tool.name,
                "description": tool.description or "",
                "input_schema": tool.inputSchema,
            }
            for tool in response.tools
        ]

    async def process_query(self, query: str) -> str:
        """处理用户查询 — MCP 工具调用循环"""
        messages = [{"role": "user", "content": query}]

        # 第一次调用 Claude
        response = self.anthropic.messages.create(
            model=self.model,
            max_tokens=4096,
            messages=messages,
            tools=self._tools_cache,
        )

        # 工具调用循环
        tool_results = []
        max_iterations = 10  # 防止无限循环

        for _ in range(max_iterations):
            # 收集 Claude 的响应
            assistant_content = response.content
            has_tool_use = False

            for block in assistant_content:
                if block.type == "text":
                    print(f"\n助手: {block.text}")
                elif block.type == "tool_use":
                    has_tool_use = True
                    print(f"\n调用工具: {block.name}({json.dumps(block.input, ensure_ascii=False)})")

                    # 调用 MCP Server 的工具
                    result = await self.session.call_tool(block.name, block.input)

                    # 提取文本结果
                    result_text = "\n".join(
                        c.text for c in result.content if hasattr(c, "text")
                    )
                    print(f"工具结果: {result_text[:200]}...")

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

            if not has_tool_use:
                break  # Claude 不再需要调用工具，结束循环

            # 将工具结果发送回 Claude
            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})

            response = self.anthropic.messages.create(
                model=self.model,
                max_tokens=4096,
                messages=messages,
                tools=self._tools_cache,
            )
            tool_results = []

        return "完成"

    async def chat_loop(self):
        """交互式对话循环"""
        print("\nMCP Agent 已就绪！输入问题开始对话，输入 'quit' 退出。")

        while True:
            query = input("\n你: ").strip()
            if query.lower() == "quit":
                break
            if not query:
                continue

            try:
                await self.process_query(query)
            except Exception as e:
                print(f"错误: {e}")

    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    agent = MCPAgent()
    try:
        server_script = sys.argv[1] if len(sys.argv) > 1 else "server.py"
        await agent.connect_to_server(server_script)
        await agent._cache_tools()
        await agent.chat_loop()
    finally:
        await agent.cleanup()


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

工具调用循环详解

上面的 Agent 使用了一个工具调用循环，这是 MCP Client 集成 LLM 的核心模式：

为什么需要循环？ 因为 Claude 可能需要多次调用工具才能回答一个问题。例如：

  1. 用户："帮我查看 config.json 并告诉我数据库连接信息"
  2. Claude 第一次 → 调用 list_directory 查找文件
  3. 获取结果 → Claude 第二次 → 调用 read_file 读取内容
  4. 获取结果 → Claude 第三次 → 直接返回分析结果

TypeScript 客户端开发

基础 TypeScript 客户端

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

class MCPClient {
  private client: Client;
  private transport: StdioClientTransport | null = null;

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

  async connectToServer(serverScript: string) {
    const isPython = serverScript.endsWith(".py");

    this.transport = new StdioClientTransport({
      command: isPython ? "python" : "node",
      args: [serverScript],
    });

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

    // 列出工具
    const toolsResult = await this.client.listTools();
    console.log(`发现 ${toolsResult.tools.length} 个工具：`);
    toolsResult.tools.forEach((t) => {
      console.log(`  - ${t.name}: ${t.description}`);
    });
  }

  async callTool(name: string, args: Record<string, unknown>) {
    const result = await this.client.callTool({ name, arguments: args });
    return result;
  }

  async listResources() {
    return await this.client.listResources();
  }

  async readResource(uri: string) {
    return await this.client.readResource({ uri });
  }

  async disconnect() {
    await this.client.close();
  }
}

// 使用示例
async function main() {
  const client = new MCPClient();
  try {
    await client.connectToServer("server.ts");

    const result = await client.callTool("read_file", { filepath: "test.txt" });
    console.log("结果:", result);
  } finally {
    await client.disconnect();
  }
}

main().catch(console.error);

配置与使用现有平台

Claude Desktop 配置

Claude Desktop 是最简单的 MCP 客户端，只需编辑配置文件：

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/user/Documents",
        "/Users/user/Desktop"
      ]
    },
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/path/to/database.db"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    },
    "custom-python": {
      "command": "python",
      "args": ["/path/to/my_server.py"],
      "env": {
        "API_KEY": "your-api-key"
      }
    }
  }
}

配置字段说明：

字段说明
command启动 Server 的命令（python、node、npx、uvx 等）
args命令参数数组
env环境变量（用于传递 API Key 等敏感信息）

Cursor IDE 配置

在 Cursor 的 Settings → MCP 中添加配置：

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
    }
  }
}

VS Code 配置

通过 VS Code 扩展（如 Claude Code 扩展）的设置添加 MCP Server 配置，格式与 Claude Desktop 一致。

常见官方 MCP Server

Server包名功能
Filesystem@modelcontextprotocol/server-filesystem读写文件，目录操作
GitHub@modelcontextprotocol/server-github仓库、Issue、PR 管理
GitLab@modelcontextprotocol/server-gitlabGitLab 项目管理
SQLitemcp-server-sqliteSQLite 数据库查询
PostgreSQL@modelcontextprotocol/server-postgresPostgreSQL 数据库查询
Brave Search@modelcontextprotocol/server-brave-searchWeb 搜索
Puppeteer@modelcontextprotocol/server-puppeteer浏览器自动化
Slack@modelcontextprotocol/server-slackSlack 消息管理
Google Drive@modelcontextprotocol/server-gdriveGoogle Drive 文件访问

快速使用示例：

# 使用 npx 运行文件系统 Server
npx -y @modelcontextprotocol/server-filesystem /path/to/dir

# 使用 uvx 运行 SQLite Server
uvx mcp-server-sqlite --db-path /path/to/db.sqlite

多 Server 集成

实际应用中，一个 Client 通常需要连接多个 Server，每个 Server 负责不同领域：

在 Claude Desktop 中配置多个 Server 非常简单，只需在 mcpServers 中添加多个条目即可。Claude 会自动将所有 Server 的工具合并供使用。

总结

要点内容
Client 角色LLM 和 Server 之间的桥梁，负责格式转换和消息路由
工具调用循环LLM 可能需要多次调用工具，Client 需要实现循环逻辑
资源管理使用 AsyncExitStack 确保连接正确关闭
平台集成Claude Desktop、Cursor、VS Code 等平台通过配置文件即可使用
多 Server一个 Agent 可以同时连接多个 Server，工具自动合并
Default footer
Copyright © 2026 HSXISAWD