从协议层出发，用 Python 实现一个完整的 MCP Server，包括 stdio 和 Streamable HTTP 两种传输方式。每个请求做什么、返回值为什么这样设计，逐一拆解。

MCP 协议基础

MCP 基于 JSON-RPC 2.0，客户端发请求，服务端回响应。三种传输层：

传输方式	原理	场景
stdio	父进程通过 stdin/stdout 收发 JSON	本地工具，Claude Desktop/Code 默认用这个
SSE（旧版）	先 GET 建立 SSE 长连接，再 POST 发请求	已被 Streamable HTTP 取代
Streamable HTTP	单个 HTTP 端点，POST 请求，响应可普通 JSON 也可升级为 SSE 流	远程服务、多客户端共享，当前推荐方案

Streamable HTTP 与旧版 SSE 的区别：旧版 SSE 要求先建立长连接再通信，Streamable HTTP 简化为单个端点（如 /mcp），无状态请求直接返回 JSON，需要流式响应时才升级为 SSE。同时通过 Mcp-Session-Id header 管理会话。

Streamable HTTP 通信模型：

客户端                          服务端
  │                               │
  │─── POST /mcp ────────────────►│  普通请求
  │◄── 200 JSON ─────────────────│  直接返回 JSON
  │                               │
  │─── POST /mcp ────────────────►│  流式请求
  │◄── 200 SSE stream ──────────│  升级为 SSE 流
  │    data: {...}                │
  │    data: {...}                │
  │                               │
  │─── GET /mcp ─────────────────►│  建立通知通道（可选）
  │◄── 200 SSE stream ──────────│  接收服务端推送

生命周期

客户端                          服务端
  │                               │
  │─── initialize ───────────────►│  能力协商
  │◄── initialize (result) ───────│
  │─── notifications/initialized ►│  通知就绪
  │                               │
  │─── tools/list ───────────────►│  正常使用阶段
  │◄── tools/list (result) ───────│
  │─── tools/call ───────────────►│
  │◄── tools/call (result) ───────│
  │         ...                   │

三个阶段：初始化 → 能力协商 → 正常通信。

---

请求详解

initialize — 握手协商

客户端发起，双方交换能力声明。

{
  "method": "initialize"
}

{
  "capabilities": {
    "experimental": {},
    "prompts": {
      "listChanged": false
    },
    "resources": {
      "subscribe": false,
      "listChanged": false
    },
    "tools": {
      "listChanged": false
    }
  },
  "serverInfo": {
    "name": "mcp-router",
    "version": "1.27.1"
  }
}

返回值设计：capabilities 声明服务端支持哪些能力（tools/resources/prompts），空对象 {} 表示支持该能力但无额外选项。客户端据此决定能调用哪些方法。

---

resources — 资源查询

{
  "method": "resources/list",
  "params": {}
}

{
  "resources": [
    {
      "name": "list_all_tools",
      "uri": "router://tools",
      "description": "列出所有服务和工具",
      "mimeType": "text/plain"
    },
    {
      "name": "read_all_resources",
      "uri": "router://resources",
      "description": "读取所有子 MCP 的资源内容。",
      "mimeType": "text/plain"
    }
  ]
}

{
  "method": "resources/read",
  "params": {
    "uri": "router://tools"
  }
}

{
  "contents": [
    {
      "uri": "router://tools",
      "mimeType": "text/plain",
      "text": "[\n  {\n    \"uri\": \"http_demo://greet\",\n    \"server\": \"http_demo\",\n    \"tool_name\": \"greet\",\n    \"description\": \"打招呼。name 为用户名。\",\n    \"params\": [\n      \"name\"\n    ]\n  },\n  {\n    \"uri\": \"http_demo://add\",\n    \"server\": \"http_demo\",\n    \"tool_name\": \"add\",\n    \"description\": \"两数相加。\",\n    \"params\": [\n      \"a\",\n      \"b\"\n    ]\n  },\n  {\n    \"uri\": \"weather://get_weather\",\n    \"server\": \"weather\",\n    \"tool_name\": \"get_weather\",\n    \"description\": \"查询指定城市的天气信息。city 为城市名称（如：北京、上海）。\",\n    \"params\": [\n      \"city\"\n    ]\n  },\n  {\n    \"uri\": \"mysql://mysql_query\",\n    \"server\": \"mysql\",\n    \"tool_name\": \"mysql_query\",\n    \"description\": \"[MySQL MCP Server [vundefined]] Run SQL queries against MySQL database (READ-ONLY)\",\n    \"params\": [\n      \"sql\"\n    ]\n  }\n]"
    }
  ]
}

---

tools/list — 声明可用工具

客户端问"你能干什么"，服务端把工具清单列出来。

{
  "method": "tools/list",
  "params": {}
}

{
  "tools": [
    {
      "name": "router_call_tool",
      "description": "调用子服务的工具。先读 router://tools 获取可用的 server_name、tool_name 和参数格式，再调用。",
      "inputSchema": {
        "type": "object",
        "properties": {
          "server_name": {
            "title": "Server Name",
            "type": "string"
          },
          "tool_name": {
            "title": "Tool Name",
            "type": "string"
          },
          "arguments": {
            "additionalProperties": true,
            "title": "Arguments",
            "type": "object"
          }
        },
        "required": [
          "server_name",
          "tool_name",
          "arguments"
        ],
        "title": "router_call_toolArguments"
      },
      "outputSchema": {
        "type": "object",
        "properties": {
          "result": {
            "title": "Result",
            "type": "string"
          }
        },
        "required": [
          "result"
        ],
        "title": "router_call_toolOutput"
      }
    }
  ]
}

返回值设计：

• inputSchema 用 JSON Schema 描述参数类型，LLM 据此构造调用参数——这是 MCP 的核心设计之一，让模型能"理解"每个工具需要什么输入
• outputSchema 用 JSON Schema 描述返回值结构，让 LLM 提前知道工具会输出什么格式的数据，便于模型决定是否调用和如何处理结果
• description 是给 LLM 看的，决定模型在什么场景下选择调用这个工具，写得好不好直接影响调用准确率

---

tools/call — 执行工具

客户端带着参数调用具体工具。

{
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "北京"
    }
  }
}

// 服务端响应（成功）
{
  "content": [
    {
      "type": "text",
      "text": "北京：晴，25°C，湿度 40%，西北风 3级"
    }
  ]
}

// 服务端响应（失败）
{
  "isError": true,
  "content": [
    {
      "type": "text",
      "text": "城市 'xxx' 不存在"
    }
  ]
}

返回值设计：

• content 是数组，支持返回多段内容（比如一段文字 + 一张图片）
• isError 不走 JSON-RPC 的 error 字段，而是放在 result 里——这样即使工具执行失败，LLM 也能拿到错误信息并自行决定下一步（重试 / 换工具 / 告知用户），而不是直接中断
• type 支持 text、image、resource，覆盖 LLM 能处理的全部内容类型

---

Python 实现（stdio）

mkdir mcp-weather-py && cd mcp-weather-py
pip install "mcp[cli]"

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")


@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的天气信息

    Args:
        city: 城市名称，如 '北京'
    """
    data = {
        "北京": "晴，25°C，湿度 40%",
        "上海": "多云，28°C，湿度 65%",
        "广州": "雷阵雨，31°C，湿度 80%",
    }
    return data.get(city, "暂无该城市数据")


@mcp.resource("cities://supported")
def city_list() -> str:
    """返回支持查询的城市列表"""
    # FastMCP 自动封装为标准结构：
    # { "contents": [{ "uri": "cities://supported", "mimeType": "text/plain", "text": "<json字符串>" }] }
    # handler 只需返回 text 内容，uri 从装饰器参数取，mimeType 默认 text/plain
    import json
    return json.dumps([
        {"name": "北京", "code": "BJ"},
        {"name": "上海", "code": "SH"},
        {"name": "广州", "code": "GZ"},
        {"name": "深圳", "code": "SZ"},
    ], ensure_ascii=False, indent=2)


if __name__ == "__main__":
    # 【传输方式选择】
    # 1. stdio：transport="stdio"（默认）—— 本地进程通信
    # 2. SSE（旧版）：transport="sse" —— 已被 Streamable HTTP 取代
    # 3. Streamable HTTP：transport="streamable-http" —— 远程部署推荐
    mcp.run()  # 默认 stdio

claude mcp add weather-py python $(pwd)/server.py

Python SDK 的设计：FastMCP 用装饰器注册工具和资源，函数签名 + docstring 自动生成 inputSchema 和 description。返回纯字符串时自动包装成 { content: [{ type: "text", text: "..." }] }。

---

Python 实现（Streamable HTTP）

# http_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-http", host="0.0.0.0", port=3000)


@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    data = {"北京": "晴，25°C", "上海": "多云，28°C"}
    return data.get(city, "暂无数据")


@mcp.resource("cities://supported")
def city_list() -> str:
    """返回支持查询的城市列表"""
    import json
    return json.dumps([
        {"name": "北京", "code": "BJ"},
        {"name": "上海", "code": "SH"},
    ], ensure_ascii=False)


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

python http_server.py
claude mcp add --transport streamable-http weather-http http://localhost:3000/mcp

Streamable HTTP vs stdio 的本质区别：stdio 是进程间通信，生命周期跟进程走；Streamable HTTP 是网络通信，需要自己管理会话（Mcp-Session-Id）、处理并发、考虑无状态场景下的请求路由。代价是灵活——可以部署到任何地方，多客户端共享。

---

Python 客户端测试

用 MCP Python SDK 写一个客户端，连接 MCP Server 并调用工具，验证整个流程是否正常。

pip install "mcp[cli]"

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


async def main():
    # 配置要连接的 MCP Server（stdio 模式）
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 1. 初始化握手
            await session.initialize()
            print("✅ 连接成功\n")

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

            # 3. 调用工具
            result = await session.call_tool("get_weather", {"city": "北京"})
            print(f"🌤️ get_weather('北京'):")
            for content in result.content:
                print(f"  {content.text}")
            print()

            # 4. 列出资源
            resources_result = await session.list_resources()
            print("📦 可用资源:")
            for res in resources_result.resources:
                print(f"  - {res.name}: {res.uri}")
            print()

            # 5. 读取资源
            read_result = await session.read_resource("cities://supported")
            print(f"📖 cities://supported:")
            for content in read_result.contents:
                print(f"  {content.text}")


asyncio.run(main())

# 确保 server.py 在同目录下
python test_client.py

预期输出：

✅ 连接成功

📋 可用工具:
  - get_weather: 获取指定城市的天气信息

🌤️ get_weather('北京'):
  晴，25°C，湿度 40%

📦 可用资源:
  - city_list: cities://supported

📖 cities://supported:
  [
    {"name": "北京", "code": "BJ"},
    {"name": "上海", "code": "SH"},
    {"name": "广州", "code": "GZ"},
    {"name": "深圳", "code": "SZ"},
  ]

---

调试技巧

手动发送 JSON-RPC 请求

# 手动向 MCP Server 发送请求测试（stdio 模式）
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}' | python server.py

MCP Inspector

官方提供的可视化调试工具，可以在浏览器中查看 MCP Server 的工具列表、资源、调用工具并查看响应。

npx @modelcontextprotocol/inspector