MCP 完全指南：协议、实战与集成

MCP（Model Context Protocol）是 2024 年 11 月由 Anthropic 主导发布并开源的协议，目标是统一 LLM 与外部工具/数据源的通信方式，让任何支持 MCP 的 LLM/Agent 都能调用任何 MCP Server 提供的 Tool，类似 “Agent Tool 的 USB-C 接口”。

目录

一、什么是 MCP

1.1 为什么需要 MCP

在 MCP 出现之前，开发者要为每个 LLM × 每个工具组合单独适配，开发量呈指数级增长：

没有 MCP：ChatGPT + GitHub、ChatGPT + Slack、Claude + GitHub、Claude + Slack ...
有 MCP：GitHub MCP Server、Slack MCP Server（一次开发，所有 LLM 通用）
角色 比喻
Host（Claude Desktop、Cursor、Cline） 插上 U 盘的电脑
Client（MCP 客户端） USB 接口
Server（提供 Tool 的服务） U 盘本身

1.2 MCP 的核心特性

  • 标准化：一次开发，所有兼容 MCP 的 Host 都能用
  • 双传输：本地进程（stdio）+ 远程服务（HTTP/SSE）
  • 三类能力：
    • Tools — 可调用的函数（最常用）
    • Resources — 可读取的数据（文件、数据库记录）
    • Prompts — 预设的提示词模板
  • 开源：协议规范、SDK、参考实现全部开源

二、MCP 协议规范

2.1 协议架构

┌─────────────────┐
│  MCP Host       │  Claude Desktop / Cursor / 自研 Agent
│  (LLM 应用)     │
└────────┬────────┘
         │ MCP Protocol (JSON-RPC 2.0)
         │
┌────────┴────────┐
│  MCP Client     │  协议客户端
└────────┬────────┘
         │
   ┌─────┴──────┬──────────┬─────────┐
   ▼            ▼          ▼         ▼
┌──────┐   ┌──────┐   ┌──────┐  ┌──────┐
│Server│   │Server│   │Server│  │Server│
│  A   │   │  B   │   │  C   │  │  D   │
└──────┘   └──────┘   └──────┘  └──────┘
 (GitHub)   (Slack)    (DB)     (Files)

2.2 通信协议

MCP 基于 JSON-RPC 2.0，所有消息分四类：

类型 方向 说明
Request Client → Server 调用方法，期待响应
Response Server → Client 请求的结果
Notification 任意方向 单向通知，无需响应
Error Server → Client 错误响应

2.3 三大核心能力

🔧 Tools（工具）

可被 LLM 调用的函数，结构完全对齐 OpenAI Function Calling：

{
  "name": "get_weather",
  "description": "查询城市天气",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": {"type": "string", "description": "城市名"}
    },
    "required": ["city"]
  }
}
📄 Resources（资源）

客户端可读取的上下文数据（文件内容、数据库记录等）：

{
  "uri": "file:///docs/readme.md",
  "name": "项目说明",
  "mimeType": "text/markdown",
  "description": "项目 README 文档"
}
💬 Prompts（提示词模板）

预设的可复用提示词：

{
  "name": "code_review",
  "description": "代码审查模板",
  "arguments": [
    {"name": "language", "description": "编程语言", "required": true}
  ]
}

2.4 生命周期（握手流程）

1. initialize         客户端发起连接，声明协议版本和能力
   ↓
2. initialized        服务端确认，握手完成
   ↓
3. tools/list         客户端列出可用工具
   ↓
4. tools/call         客户端发起工具调用
   ↓
5. tools/call 响应    服务端返回结果

2.5 传输方式

2.5.1 三种传输方式概览
方式 适用场景 特点
stdio 本地进程集成 最常用，最简单
HTTP + SSE 远程服务 支持分布式部署
WebSocket 实验性 双向实时通信
2.5.2 深入理解 stdio（推荐重点掌握）

stdio = Standard Input/Output（标准输入/输出），是操作系统为每个进程提供的 3 个默认数据流：

流 名称 默认设备 文件描述符
stdin 标准输入 键盘 fd = 0
stdout 标准输出 终端屏幕 fd = 1
stderr 标准错误 终端屏幕 fd = 2 
        进程
       ┌──────────┐
键盘 ─►│ stdin  0 │──► 进程内部
       │          │
       │ stdout 1 │──► 屏幕（程序输出）
       │          │
       │ stderr 2 │──► 屏幕（错误信息）
       └──────────┘

stdio 的核心能力：重定向与管道

普通程序运行时，stdin/stdout/stderr 自动绑定到当前终端，但可以重定向到文件或其他程序：

# 输出重定向到文件
python hello.py > output.txt

# 输入重定向到文件
python hello.py < input.txt

# 错误重定向
python hello.py 2> error.log

# 管道：把 A 的 stdout 接到 B 的 stdin
cat file.txt | grep "error" | wc -l

这就是 Unix 哲学的基石：“每个程序只做一件事，通过 stdin/stdout 与其他程序协作”。

2.5.3 stdio 在 MCP 中的工作原理

MCP 的 stdio 传输就是利用 stdio 的进程间通信（IPC） 能力：

┌──────────────┐                    ┌──────────────────┐
│  MCP Host    │                    │   MCP Server     │
│  (Claude)    │                    │  (你的 Python)   │
│              │                    │                  │
│  ┌────────┐  │  stdin  ◄────────  │  stdout          │
│  │ Client │──┼──────────►         │                  │
│  └────────┘  │  stdout ────────► │  stdin           │
│              │                    │                  │
└──────────────┘                    └──────────────────┘
        Host 把 Server 作为子进程启动，通过 stdio 收发 JSON-RPC 消息

启动流程：

1. Claude Desktop 执行 command（python weather_server.py）
2. 启动一个子进程（Python 进程）
3. Claude Desktop 接管这个子进程的 stdin/stdout
4. 通过 stdin 发送 JSON-RPC 请求
5. 从 stdout 接收 JSON-RPC 响应
6. stderr 用于日志/错误信息（不会被协议解析）
2.5.4 stdio MCP Server 实现示例
from mcp.server import Server
from mcp.server.stdio import stdio_server   # ⭐ stdio 传输
from mcp.types import Tool, TextContent
import asyncio

app = Server("weather-server")

@app.list_tools()
async def list_tools():
    return [Tool(name="get_weather", description="查天气", inputSchema={...})]

@app.call_tool()
async def call_tool(name, arguments):
    return [TextContent(type="text", text="晴天 25°C")]

async def main():
    # ⭐ 关键：stdio_server() 把 stdin/stdout 包装成异步流
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream,
                     app.create_initialization_options())

asyncio.run(main())

stdio_server() 内部实现：

# 简化版源码
async def stdio_server():
    # 把 sys.stdin 包装成可读异步流
    read_stream = anyio.wrap_file(sys.stdin)
    # 把 sys.stdout 包装成可写异步流
    write_stream = anyio.wrap_file(sys.stdout)
    return read_stream, write_stream
2.5.5 stdio vs HTTP+SSE 对比
维度 stdio（本地） HTTP + SSE（远程）
部署位置 本地机器 远程服务器
进程模型 子进程（每个客户端一个进程） 常驻服务（多客户端共享）
通信方式 stdin/stdout HTTP 请求 + SSE 推送
启动方式 command + args URL + headers
优点 简单、零网络配置、安全（数据不出本机） 跨网络、可扩展、共享资源
缺点 不能跨机器、性能受进程开销影响 需要部署、需要鉴权、网络延迟
典型场景 Claude Desktop / Cursor 本地工具 企业内网共享服务

stdio 配置示例：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    }
  }
}

HTTP+SSE 配置示例（远程 MCP Server）：

{
  "mcpServers": {
    "remote-weather": {
      "url": "https://mcp.weather.example.com/sse",
      "headers": {
        "Authorization": "Bearer your-api-token"
      }
    }
  }
}
2.5.6 stdio 在 MCP 中流行的 5 个原因
  1. 零网络配置 —— 不需要端口、域名、TLS 证书
  2. 天然安全 —— 数据不离开本机，适合访问本地资源（文件、数据库）
  3. 进程隔离 —— Server 崩溃不会影响 Host
  4. 易于开发调试 —— 直接命令行运行即可看到 stdout 输出
  5. 跨平台 —— Windows/Mac/Linux 行为一致

这就是为什么 90% 的 MCP Server 都优先支持 stdio 传输。

2.5.7 动手验证 stdio

1. 体验管道（stdio 重定向）

$ echo "hello world" | grep "hello"
# echo 把 "hello world" 写到自己的 stdout
# grep 从自己的 stdin 读取
# 两者通过管道连接

2. 写一个最简单的 stdio 程序

# echo_stdio.py
import sys

# 从 stdin 读一行
line = sys.stdin.readline().strip()
# 把处理结果写到 stdout
sys.stdout.write(f"你说: {line}\n")
sys.stdout.flush()   # 重要！stdio 通信必须 flush

# 测试：键盘输入
$ python echo_stdio.py
hello            # 键盘输入 → stdin
你说: hello      # stdout 输出

# 测试：通过管道
$ echo "hi" | python echo_stdio.py
你说: hi

3. MCP Server 本地调试

# 直接命令行启动，stdin 接键盘
$ python weather_server.py

# 或者用 MCP Inspector 调试（自动接管 stdio）
$ npx @modelcontextprotocol/inspector python weather_server.py

Inspector 会自动接管 stdin/stdout，你可以可视化测试工具调用。

2.5.8 深入理解 IPC（进程间通信）

上一节提到 stdio 传输利用了"进程间通信（IPC）"能力，这里系统讲解一下 IPC 的概念与原理。

一、什么是 IPC

IPC（Inter-Process Communication） 指不同进程之间传递数据或信号的机制。

每个进程都有独立的内存空间（虚拟地址空间），不能直接访问对方的数据，必须通过操作系统提供的"通道"才能交流。

┌─────────────────┐   ┌─────────────────┐
│   进程 A        │   │   进程 B        │
│  ┌───────────┐  │   │  ┌───────────┐  │
│  │ 用户代码  │  │   │  │ 用户代码  │  │
│  └─────┬─────┘  │   │  └─────┬─────┘  │
│  ┌─────▼─────┐  │   │  ┌─────▼─────┐  │
│  │ 共享库    │  │   │  │ 共享库    │  │
│  └─────┬─────┘  │   │  └─────┬─────┘  │
│  ┌─────▼─────┐  │   │  ┌─────▼─────┐  │
│  │ 堆/栈     │  │   │  │ 堆/栈     │  │
│  └───────────┘  │   │  └───────────┘  │
│   0x0000~0xFFFF │   │   0x0000~0xFFFF │  ← 各自独立
└─────────────────┘   └─────────────────┘
        │                      │
        └──────────┬───────────┘
                   ▼
          ┌──────────────────────┐
          │   操作系统内核        │
          │  （唯一可以共享区域） │
          └──────────────────────┘

关键点：进程 A 不能直接读进程 B 的内存，必须经由内核作为中介。

二、IPC 的 7 种主要方式
方式 通信方向 是否跨机器 典型用途
管道（Pipe） 单向 否 命令行管道、父子进程
命名管道（FIFO） 单向 否 任意进程间
信号（Signal） 单向通知 否 Ctrl+C、kill
消息队列（Message Queue） 双向 否 系统 V IPC
共享内存（Shared Memory） 双向 否 高性能场景
信号量（Semaphore） 同步 否 进程同步
套接字（Socket） 双向 ✅ 网络、跨机器
文件 双向 ✅（共享文件系统） 简单场景

本地 IPC vs 网络 IPC：

维度 本地 IPC 网络 IPC
范围 同一台机器 跨机器
速度 极快（微秒级） 较慢（毫秒级，受网络影响）
协议 管道、共享内存、信号 TCP/UDP、HTTP、WebSocket
鉴权 OS 权限 网络鉴权、加密
典型场景 本地工具链 微服务、分布式系统

stdio 属于本地 IPC 中的"管道"方式。

三、stdio 属于哪种 IPC

stdio 是 匿名管道（Anonymous Pipe） 的一种特殊形式。

维度 匿名管道（Pipe） 命名管道（FIFO）
标识 无名字（内核管理） 文件系统中有名字
范围 只能父子/兄弟进程 任意进程
生命周期 进程结束自动销毁 文件系统存在期间
典型用途 ls | grep 客户端-服务进程通信
在 MCP 中的角色 ✅ stdio 用这种 ❌ 不用

stdio 为什么是匿名管道：

当 Claude Desktop 执行 python weather_server.py 时：

1. Claude Desktop 进程（父进程）
2. fork() 创建子进程（Python 进程）
3. 子进程继承父进程的 stdin/stdout
4. 父进程把管道的一端接到子进程的 stdin/stdout
5. 父子进程通过这条管道通信

这就是典型的匿名管道场景：父子进程之间。

四、管道（Pipe）的工作原理

管道本质上是内核中的一块缓冲区：

┌──────────┐         内核缓冲区          ┌──────────┐
│ 进程 A   │──write()──►│              │◄──read()──│ 进程 B   │
│ (写)     │            │   4KB~64KB   │           │ (读)     │
└──────────┘            │              │           └──────────┘
         ◄─── 半双工 ───►│              │
                        └──────────────┘

关键特性：

特性 说明
半双工 数据单向流动（A→B 或 B→A）
全双工变种 stdio 用 stdin/stdout 各一条管道，相当于全双工
阻塞 缓冲区空时 read() 阻塞；缓冲区满时 write() 阻塞
字节流 没有消息边界，需要自己定义协议（JSON-RPC 用换行符分隔）
自动销毁 所有引用该管道的 fd 关闭后自动释放
五、MCP 用 stdio 的本质

MCP 协议在 stdio 上做了三层包装：

┌────────────────────────────────────────────┐
│  Layer 2：MCP 协议（JSON-RPC 2.0）          │
│  - initialize / tools/list / tools/call    │
├────────────────────────────────────────────┤
│  Layer 1：stdio 上的消息定界                │
│  - 每条 JSON-RPC 消息以 \n 结尾            │
│  - Content-Length 头（可选，类似 LSP 协议） │
├────────────────────────────────────────────┤
│  Layer 0：匿名管道（stdin/stdout）          │
│  - 4KB 内核缓冲区                          │
│  - 半双工 → 用两条管道模拟全双工            │
└────────────────────────────────────────────┘
六、消息定界问题

管道是字节流，没有消息边界：

# 写端发了 3 条消息
echo "msg1"; echo "msg2"; echo "msg3"

# 读端可能一次性收到：
"msg1\nmsg2\nmsg3\n"

# 也可能分两次收到：
"msg1\nmsg2\n"
"msg3\n"

MCP 的解决方案 —— 用 LSP 风格 的消息头：

Content-Length: 87\r\n
\r\n
{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}

或者更简单的换行分隔（stdio 模式默认）：

{"jsonrpc":"2.0","id":1,"method":"initialize",...}\n
{"jsonrpc":"2.0","id":2,"method":"tools/list"}\n
七、stdio 在 MCP 中的完整数据流

单次请求-响应流程：

Host (Claude Desktop)                        Server (Python)
      │                                            │
      │  1. stdin: {"method":"tools/call",...}    │
      │ ──────────────────────────────────────────►│
      │                                            │  解析 JSON
      │                                            │  执行工具
      │                                            │  构造响应
      │  2. stdout: {"result":"晴天 25°C"}         │
      │ ◄──────────────────────────────────────────│
      │                                            │
      3. 解析响应，交给 LLM 处理

实际看到的字节流（简化版）：

Host 写入 Server stdin：

{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18",...}}
{"jsonrpc":"2.0","method":"notifications/initialized"}
{"jsonrpc":"2.0","id":2,"method":"tools/list"}
{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"get_weather","arguments":{"city":"北京"}}}

Server 从 stdout 写回：

{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{"tools":{}},...}}
{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"get_weather","description":"...","inputSchema":{...}}]}}
{"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"北京：晴，温度 25°C"}]}}

所有 stderr 输出（print("xxx")）都不会被协议解析，可用于日志调试。

八、其他 IPC 方式在 MCP 生态中的角色
IPC 方式 在 MCP 中的应用
stdio（匿名管道） 本地 Server，90% 场景
Socket（TCP/HTTP+SSE） 远程 Server，云服务
WebSocket 远程 Server，双向实时
共享内存 ❌ 不用于 MCP（太低层）
消息队列 ❌ 不用于 MCP（太重）
命名管道 Windows 上偶尔用于跨进程
九、动手观察 IPC

1. 观察 stdio 的 IPC：

# 启动 Python 子进程，往它的 stdin 写 JSON，看 stdout 输出
$ echo '{"method":"tools/list","id":1}' | python weather_server.py
# 父 shell 进程 → Python 子进程（通过管道）
# Python 把响应写到 stdout → 父 shell 显示出来

2. 用 strace 观察系统调用（Linux）：

$ strace -e trace=read,write python weather_server.py
read(0, "{\"method\":\"tools/list\"...}", 4096) = 35
write(1, "{\"jsonrpc\":\"2.0\",\"id\":1,...}", 87) = 87
#                ↑
#           fd=1 是 stdout

这就是 IPC 在系统层面的真实样子：read(0, ...) 和 write(1, ...) 系统调用。

3. Windows 观察：

Windows 没有 strace，但可以用 Process Monitor（Sysinternals 工具）观察：

  • 进程创建事件
  • 文件句柄读写（stdin/stdout 也是文件句柄）
十、总结
概念 要点
IPC 不同进程间传递数据/信号的机制
stdio 属于 本地 IPC 中的"匿名管道"方式
为什么用 stdio 简单、零配置、安全（数据不出本机）
MCP 协议层 JSON-RPC 2.0 消息 + 换行/Content-Length 定界 + stdio 字节流
其他 IPC Socket（远程）、共享内存（高性能）、消息队列（重型）
本质 父子进程通过内核缓冲区传递字节

一句话总结：MCP 的 stdio 传输 = 匿名管道（IPC 的一种） + JSON-RPC 2.0 消息协议。

三、怎么接入 MCP（4 种方式）

方式 1：使用 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", "C:/Users/me/docs"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {"GITHUB_TOKEN": "ghp_xxx"}
    }
  }
}

重启 Claude Desktop，工具即生效。

方式 2：使用 Cursor 编辑器

Settings → MCP → Add new global MCP server，配置同上。

方式 3：开发自己的 MCP Client（Python）

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

async def main():
    # 启动 MCP Server 子进程
    server_params = StdioServerParameters(
        command="python",
        args=["weather_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 初始化
            await session.initialize()

            # 2. 列出工具
            tools = await session.list_tools()
            print("可用工具：", [t.name for t in tools.tools])

            # 3. 调用工具
            result = await session.call_tool(
                "get_weather",
                arguments={"city": "北京"}
            )
            print("天气：", result.content)

asyncio.run(main())

方式 4：在 Agent 框架中接入

LangChain（Python）
from langchain_mcp import MCPToolkit

toolkit = MCPToolkit.from_stdio_server(
    command="python",
    args=["weather_server.py"]
)
tools = toolkit.get_tools()
Spring AI（Java）
McpClient mcpClient = McpClient.builder()
    .stdio("python", "weather_server.py")
    .build();
List<Tool> tools = mcpClient.listTools();

四、MCP 实战

4.1 环境准备

# Python
pip install mcp

# Node.js
npm install @modelcontextprotocol/sdk

4.2 实战 1：天气查询 MCP Server（Python）

完整代码 weather_server.py：

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncio

app = Server("weather-server")

# ===== 1. 注册工具 =====
@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="查询指定城市的实时天气情况。\n"
                       "当用户问到天气、温度、是否下雨时使用。",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市中文名，例如：北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认为 celsius"
                    }
                },
                "required": ["city"],
                "additionalProperties": False
            }
        ),
        Tool(
            name="get_forecast",
            description="查询未来 N 天的天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"},
                    "days": {
                        "type": "integer",
                        "description": "预报天数，1-7 天",
                        "minimum": 1,
                        "maximum": 7
                    }
                },
                "required": ["city", "days"]
            }
        )
    ]

# ===== 2. 实现工具逻辑 =====
def fetch_weather(city: str, unit: str = "celsius") -> dict:
    """调用真实天气 API（这里用 mock）"""
    return {
        "city": city,
        "temperature": 25 if unit == "celsius" else 77,
        "condition": "晴",
        "humidity": 60,
        "wind": "东南风 3 级"
    }

def fetch_forecast(city: str, days: int) -> list:
    """获取天气预报"""
    return [
        {"date": f"2025-08-{i+1:02d}", "high": 28, "low": 20, "condition": "晴"}
        for i in range(days)
    ]

# ===== 3. 处理工具调用 =====
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "get_weather":
            city = arguments["city"]
            unit = arguments.get("unit", "celsius")
            data = fetch_weather(city, unit)
            text = (f"{data['city']}：{data['condition']}，"
                    f"温度 {data['temperature']}°"
                    f"{'C' if unit == 'celsius' else 'F'}，"
                    f"湿度 {data['humidity']}%，{data['wind']}")
            return [TextContent(type="text", text=text)]

        elif name == "get_forecast":
            forecast = fetch_forecast(arguments["city"], arguments["days"])
            lines = [f"{f['date']}: {f['condition']}, {f['low']}~{f['high']}°C"
                     for f in forecast]
            return [TextContent(type="text", text="\n".join(lines))]

        else:
            raise ValueError(f"未知工具: {name}")

    except Exception as e:
        return [TextContent(type="text", text=f"错误: {str(e)}")]

# ===== 4. 启动服务 =====
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

4.3 实战 2：接入 Claude Desktop

步骤 1：注册到 Claude Desktop 配置

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["D:/AI-Study/study/weather_server.py"]
    }
  }
}

步骤 2：重启 Claude Desktop

步骤 3：测试

在 Claude Desktop 中输入：

帮我查一下北京今天的天气，顺便看看未来 3 天的预报

Claude 会自动调用 get_weather 和 get_forecast 工具。

4.4 实战 3：集成真实数据库（SQLite）

import sqlite3
import json
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
import asyncio

app = Server("db-server")
DB_PATH = "data.db"

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_users",
            description="查询用户信息。当用户问到用户、会员、注册时使用。",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {
                        "type": "string",
                        "description": "用户ID，例如：U10086"
                    },
                    "username": {
                        "type": "string",
                        "description": "用户名（模糊匹配），可选"
                    }
                }
            }
        ),
        Tool(
            name="list_tables",
            description="列出数据库中所有表。当用户询问数据库结构时使用。",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    try:
        if name == "query_users":
            user_id = arguments.get("user_id")
            username = arguments.get("username")

            sql = "SELECT id, username, email, created_at FROM users WHERE 1=1"
            params = []
            if user_id:
                sql += " AND id = ?"
                params.append(user_id)
            if username:
                sql += " AND username LIKE ?"
                params.append(f"%{username}%")

            cursor.execute(sql, params)
            rows = cursor.fetchall()
            result = [{"id": r[0], "username": r[1], "email": r[2], "created_at": r[3]}
                      for r in rows]
            return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]

        elif name == "list_tables":
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
            tables = [row[0] for row in cursor.fetchall()]
            return [TextContent(type="text", text="、".join(tables))]

        else:
            return [TextContent(type="text", text=f"未知工具: {name}")]

    except Exception as e:
        return [TextContent(type="text", text=f"查询失败: {e}")]
    finally:
        conn.close()

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

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

4.5 实战 4：Node.js 版 MCP Server

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "weather-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "get_weather",
    description: "查询城市天气",
    inputSchema: {
      type: "object",
      properties: {
        city: { type: "string", description: "城市名" }
      },
      required: ["city"]
    }
  }]
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "get_weather") {
    const { city } = request.params.arguments;
    return {
      content: [{ type: "text", text: `${city} 晴天 25°C` }]
    };
  }
  throw new Error("Unknown tool");
});

const transport = new StdioServerTransport();
await server.connect(transport);

4.6 实战 5：Java 版 MCP Server

import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpSyncServer;
import io.modelcontextprotocol.server.transport.StdioServerTransport;
import io.modelcontextprotocol.spec.McpSchema;

public class WeatherMcpServer {

    public static void main(String[] args) {
        McpSyncServer server = McpServer.sync(new StdioServerTransport())
            .serverInfo("weather-server", "1.0.0")
            .capabilities(McpSchema.ServerCapabilities.builder()
                .tools(true)
                .build())
            .tool(McpSchema.Tool.builder()
                .name("get_weather")
                .description("查询城市天气")
                .inputSchema(/* JSON Schema */)
                .build())
            .build();

        // 注册工具处理逻辑
        server.addToolHandler("get_weather", (exchange, args) -> {
            String city = (String) args.get("city");
            return new McpSchema.CallToolResult(
                List.of(new McpSchema.TextContent(city + " 晴天 25°C")),
                false
            );
        });
    }
}

五、调试与可视化工具

5.1 MCP Inspector（官方调试器）

npx @modelcontextprotocol/inspector python weather_server.py

浏览器打开 http://localhost:5173，可可视化测试工具调用。

5.2 日志调试

import logging
logging.basicConfig(level=logging.DEBUG)

5.3 常见错误排查

错误 原因 解决
Connection refused stdio 配置错误 检查 command 和 args
Tool not found 工具名拼写错误 检查 @list_tools 注册名
Invalid arguments 参数 schema 不匹配 检查 inputSchema 与 call_tool 参数
JSON-RPC parse error 协议版本不匹配 更新 SDK 到最新版

六、生态现状与主流 MCP Server

⚠️ 生态现状说明（必读）：
MCP 生态演进非常快——截至 2026 年初，社区已收录 1200+ MCP Server，但其中：

  • 约 30% 处于"个人实验"阶段，长期未维护
  • 约 40% 是社区小项目，覆盖长尾场景
  • 真正"主流稳定" 的不到 50 个

因此本章不罗列全部 Server，而是先讲清楚分类与现状，再给出真正在生产中被广泛使用的清单与深度案例。

6.1 生态结构分类

按维护方与稳定性，MCP Server 可分为三个层级：

层级 数量 特征 适用
L1 官方参考实现 ~15 个 Anthropic/Claude Desktop 官方维护，文档齐全 学习参考、生产首选
L2 厂商官方 Server ~80 个 由 GitHub、AWS、Notion、Stripe 等公司官方维护 生产可用
L3 社区 Server 1000+ 个人或小团队维护，质量参差 评估后试用

6.2 主流 MCP Server（实战真正在用）

以下清单综合 GitHub Star、官方文档收录、Claude Desktop Connectors 默认安装、生产使用反馈 等维度筛选：

🏆 L1：官方参考实现（Anthropic 维护）
Server 标识 功能 安装命令
Filesystem fs 读写本地文件/目录 npx -y @modelcontextprotocol/server-filesystem <path>
GitHub github GitHub 全套 API npx -y @modelcontextprotocol/server-github
GitLab gitlab GitLab API npx -y @modelcontextprotocol/server-gitlab
Google Drive gdrive Google Drive 文件 npx -y @modelcontextprotocol/server-google-drive
Slack slack Slack 消息 npx -y @modelcontextprotocol/server-slack
PostgreSQL postgres PG 数据库 npx -y @modelcontextprotocol/server-postgres
SQLite sqlite SQLite 数据库 npx -y @modelcontextprotocol/server-sqlite
Redis redis Redis 缓存 npx -y @modelcontextprotocol/server-redis
Puppeteer puppeteer 浏览器自动化 npx -y @modelcontextprotocol/server-puppeteer
Fetch fetch HTTP 请求 npx -y @modelcontextprotocol/server-fetch
🏢 L2：厂商官方 Server（生产可用）
Server 厂商 特色场景 是否需要付费
GitHub MCP Server ⭐ GitHub 官方 PR / Issue / Code Review 免费（Token 即可）
Notion MCP Server Notion 官方 知识库管理 免费
Linear MCP Server Linear 官方 项目管理 免费
Slack MCP Server Slack 官方 团队通知 免费
Stripe MCP Server Stripe 官方 支付查询 免费
Sentry MCP Server Sentry 官方 错误监控排查 免费（个人版可用）
AWS MCP Servers ⭐ AWS Labs 多服务（K8s/Cost/CCAPI） 免费
Atlassian MCP Server Atlassian 官方 Jira/Confluence 部分付费
Cloudflare MCP Server Cloudflare 官方 Workers/KV/R2 免费
Supabase MCP Server Supabase 官方 DB/Auth/Storage 一站式 免费
Vercel MCP Server Vercel 官方 部署管理 免费
Asana MCP Server Asana 官方 任务管理 免费
Figma MCP Server ⭐ Figma 官方 设计转代码 付费（需 Pro+）
Context7 MCP Server ⭐ Upstash 实时文档检索（Context7.com） 免费
🌟 L2.5：社区明星 Server（非官方但极流行）
Server 维护方 特色 Star
Desktop Commander ⭐ @wonderwhy-er OS 级控制：终端、文件、进程 极高
Playwright MCP Server executeautomation 跨浏览器自动化 6.1k+
Bright Data MCP ⭐ Bright Data 反爬 + 代理 + 浏览器自动化 高
Tavily MCP Server Tavily AI 优化搜索 高
Firecrawl MCP Server Mendable 网页转 Markdown 高
Blender MCP Server 社区 3D 场景生成 中
Whisper MCP Server 社区 语音转文字 中
MongoDB MCP Server 社区 MongoDB 查询 中
ClickHouse MCP Server 社区 OLAP 分析 中
Replicate MCP Server 社区 调用 Replicate AI 模型 中
Zapier MCP Zapier 官方 连接万种 SaaS 高

📌 说明：以上清单基于 2025-2026 年社区使用数据整理，是 Claude Desktop / Cursor / Cline 等主机中实际被频繁安装的 Server。

6.3 深度案例：GitHub MCP Server

为什么把 GitHub 单独作为深度案例？因为它是开发者使用频率最高 + 工具最完整 + 官方维护最稳 的代表，几乎所有"AI Agent + 真实业务"的演示都会用到它。

6.3.1 它能做什么

GitHub MCP Server 把 GitHub 全部核心 API 封装为 Tool，覆盖日常开发的 90% 场景：

工具类别 典型工具 用途
仓库管理 create_repository、fork_repository、search_repositories 创建/搜索/分支仓库
Issue 操作 list_issues、create_issue、update_issue、add_issue_comment 工单管理
PR 管理 list_pull_requests、create_pull_request、merge_pull_request 代码评审
代码浏览 get_file_contents、search_code、list_commits 阅读代码
工作流 list_workflows、run_workflow、get_workflow_run CI/CD
通知 list_notifications、mark_notification_read 消息提醒
6.3.2 安装与配置

在 Claude Desktop 中配置：

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxx"
      }
    }
  }
}

需要 GitHub Token，权限范围：

  • repo（仓库读写）
  • issues（Issue 读写）
  • pull_requests（PR 读写）
  • actions（可选，工作流）
  • notifications（可选）
6.3.3 实战用例

用例 1：自动 Code Review

用户：帮我看看 microsoft/vscode 最近 5 个 PR 的状态，
      有没有需要我关注的？

Claude 自动调用：

  1. search_repositories 找到 microsoft/vscode
  2. list_pull_requests 获取最近 PR
  3. get_file_contents 查看具体变更
  4. 综合分析后给出结构化回复

用例 2：批量管理 Issue

用户：列出所有标签是 "bug" 且未分配的 issue，按创建时间排序

Claude 自动调用：

  1. list_issues 带 labels:["bug"]、assignee:"none"、sort:"created"
  2. 格式化输出表格

用例 3：自动修复流水线

用户：我的 CI 失败了，帮我看下日志

Claude 自动调用：

  1. list_workflow_runs 找到失败的那次
  2. get_workflow_run_logs 拉取日志
  3. 分析错误，定位代码位置
  4. 调用 get_file_contents 查看问题代码
  5. 提出修复方案
6.3.4 核心工具源码剖析

以 create_issue 为例，看 GitHub MCP Server 是怎么设计 Tool 的：

# 简化版源码
@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="create_issue",
            description="在指定 GitHub 仓库创建 Issue。\n"
                       "当用户要求报告 bug、提出功能请求、记录任务时使用。\n"
                       "可指定标签、负责人、里程碑。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {
                        "type": "string",
                        "description": "仓库所有者（用户或组织名）"
                    },
                    "repo": {
                        "type": "string",
                        "description": "仓库名"
                    },
                    "title": {
                        "type": "string",
                        "description": "Issue 标题，简明扼要"
                    },
                    "body": {
                        "type": "string",
                        "description": "Issue 详细描述，支持 Markdown"
                    },
                    "labels": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "标签列表，例如：['bug', 'urgent']"
                    },
                    "assignees": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "负责人用户名列表"
                    }
                },
                "required": ["owner", "repo", "title"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    if name == "create_issue":
        # 调用 GitHub REST API
        url = f"https://api.github.com/repos/{arguments['owner']}/{arguments['repo']}/issues"
        headers = {
            "Authorization": f"Bearer {GITHUB_TOKEN}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28"
        }
        resp = requests.post(url, json=arguments, headers=headers)
        if resp.status_code == 201:
            data = resp.json()
            return [TextContent(
                type="text",
                text=f"Issue 创建成功：{data['html_url']}\n编号 #{data['number']}"
            )]
        else:
            return [TextContent(
                type="text",
                text=f"创建失败：{resp.status_code} - {resp.text}"
            )]
6.3.5 设计亮点
  1. Tool 粒度细：每个 API 端点一个 Tool，避免"万能 Tool"
  2. description 写到位：明确说明何时使用，且给出参数示例
  3. 必填参数少：只强制 owner/repo/title，其他可选
  4. 错误信息友好：返回结构化文本而不是抛异常
  5. Markdown 支持：body 字段明确支持 Markdown
6.3.6 替代方案对比
方案 优点 缺点
GitHub MCP Server 官方、完整、稳定 需要 Token
自定义 GitHub Tool 灵活、可裁剪 自己写、需维护
LangChain GitHub Toolkit 与 LangChain 集成好 锁框架
直接用 GitHub CLI 简单 LLM 难直接调用
6.3.7 为什么 GitHub MCP Server 是学习典范
维度 表现
覆盖度 几乎 100% 覆盖日常开发场景
官方维护 GitHub 官方团队持续更新
稳定性 错误处理完善，超时控制到位
安全性 Token 通过 env 注入，不硬编码
生态成熟 文档、示例、社区齐全
可学习性 源码是学习 MCP Server 开发的最佳教材

6.4 兼容 MCP 的 Host（客户端）

Host 类型 是否原生支持
Claude Desktop 桌面应用 ✅（首个官方）
Claude Code CLI ✅
Cursor IDE ✅
Windsurf IDE ✅
Cline VS Code 插件 ✅
Continue VS Code/JetBrains 插件 ✅
Zed 编辑器 ✅
Codex CLI OpenAI CLI ✅
ChatGPT Desktop App 桌面应用 ✅
OpenAI Agents SDK SDK ✅
Amazon Bedrock AgentCore Gateway 云服务 ✅
LibreChat 开源聊天 UI ✅
Chainlit 开源聊天框架 ✅
Cherry Studio 桌面客户端 ✅
NextChat 开源聊天 UI ✅

6.5 参考链接

  • 协议规范：https://modelcontextprotocol.io
  • GitHub：https://github.com/modelcontextprotocol
  • 官方 Servers 仓库：https://github.com/modelcontextprotocol/servers
  • Awesome MCP Servers 合集：https://github.com/punkpeye/awesome-mcp-servers
  • MCP 注册中心：https://mcp.so
  • 生态总览：https://mcp-awesome.com
  • 官方 2026 生态参考：https://hidekazu-konishi.com/entry/mcp_server_ecosystem_reference_2026.html
  • Python SDK：pip install mcp
  • TypeScript SDK：npm install @modelcontextprotocol/sdk
  • Java SDK：io.modelcontextprotocol:mcp

七、实战最佳实践

7.1 描述要清晰

# ❌ 不好
description="天气工具"

# ✅ 好
description="查询指定城市的实时天气，包括温度、天气状况、风力。"
            "当用户问到天气、温度、是否下雨时使用。"

7.2 异常返回字符串

@app.call_tool()
async def call_tool(name, arguments):
    try:
        # 业务逻辑
        return [TextContent(type="text", text=result)]
    except Exception as e:
        # ✅ 返回错误给 LLM，让它组织友好回复
        return [TextContent(type="text", text=f"调用失败：{e}")]

7.3 敏感操作加确认

Tool(
    name="delete_user",
    description="删除用户。危险操作，需 confirm=True 才执行。",
    inputSchema={
        "properties": {
            "user_id": {"type": "string"},
            "confirm": {"type": "boolean", "description": "二次确认"}
        },
        "required": ["user_id", "confirm"]
    }
)

7.4 提供 Resource 减少 Tool 调用

@app.list_resources()
async def list_resources():
    return [
        Resource(
            uri="config://app",
            name="应用配置",
            mimeType="application/json"
        )
    ]

7.5 性能优化

  • 缓存：相同参数的调用结果可加缓存
  • 批量：设计 batch_xxx 工具，一次调用处理多个
  • 流式：大数据返回考虑用 SSE 流式输出
  • 超时：避免 Tool 永久阻塞整个 Agent

八、总结

维度 要点
是什么 Anthropic 主导的 LLM-工具通信协议
解决什么 避免重复适配，让 Tool 一次开发多 LLM 通用
三大能力 Tools（函数）/ Resources（数据）/ Prompts（模板）
传输方式 stdio（本地）+ HTTP+SSE（远程）
协议基础 JSON-RPC 2.0
怎么接入 配置文件（零代码）/ 写 Client / 框架集成
实战 装饰器 + Tool schema + call_tool 处理函数
未来 正在成为 Agent Tool 生态的事实标准

相关文档

# 人工智能
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

cover

API网关的作用？MAI Gateway带你快速理解AI网关核心概念

avatar MCP技术社区

企业AI基础设施架构的路径选择：共享底座与联邦互通的长期权衡

本文系统分析了企业AI基础设施的两条演进路线——联邦互通与共享运行时底座。联邦互通通过MCP等协议实现Agent间标准化连接，适合中小规模、快速试错场景；共享底座则通过统一执行层实现全流程可观测、可审计、可管控，适用于大规模、强合规的企业级部署。文章深入探讨了两者的核心分歧、共享底座的六层架构设计、关键设计取舍、成本收益分析，并提供了渐进式落地路径和决策框架，指出两种模式将长期共存、互补演进，企业

avatar MCP技术社区
cover

使用ChatGPT登录到Codex卡住？2026年6种排查与解决方法

avatar MCP技术社区