1. MCP 是什么

MCP（Model Context Protocol）可以理解为：给大模型与外部工具之间定义了一套“统一插座标准”。

以前每接一个工具常常要写不同的适配代码；有了 MCP，模型端和工具端只要都遵循这个协议，就能统一方式通信。

一句话定位：MCP = 让模型更标准地“看见并调用”外部能力（工具 / 资源 / 提示模板）的协议层。

1.1 三层架构

本项目把这套心智模型落地成代码层：决策层（LLM） / 编排层（MCP Client） / 执行层（MCP Server）。

2. 源码逻辑实现

下面按文件路径顺序讲，每个文件都配关键代码。完整代码见仓库内文件本身。

2.1 入口与配置（main.py / config.py）

main.py：交互主循环

它读用户输入 → 让 ask_with_llm 走一遍 → 打印回复。一段 async with mcp_client 进会话，等同同时完成 MCP 握手。

from __future__ import annotations
import asyncio

from client.llm import ask_with_llm, create_openai_client
from client.runtime import create_mcp_client
from config import AppConfig


async def run_chat(config: AppConfig) -> None:
    llm_client = create_openai_client(config)
    mcp_client = create_mcp_client(config)
    async with mcp_client as client:           # 进入会话 = initialize + initialized
        print("输入问题开始对话，输入 exit 退出。")
        while True:
            user_prompt = input("你：").strip()
            if user_prompt == "exit":
                return
            answer = await ask_with_llm(client, llm_client, config, user_prompt)
            print(f"助手：{answer}")


def main() -> None:
    asyncio.run(run_chat(AppConfig.from_env()))

config.py：统一配置 schema

把所有可配置项放在一个 pydantic 模型里，统一从 .env 读取，避免散落的 os.getenv。

TransportType = Literal["stdio", "sse", "streamable_http"]

class AppConfig(BaseModel):
    openai_api_key: str
    openai_base_url: str
    openai_model: str
    mcp_transport: TransportType      # 三选一
    mcp_host: str
    mcp_port: int
    mcp_sse_path: str
    mcp_streamable_path: str
    llm_max_tool_rounds: int           # 防止无线循环

    @classmethod
    def from_env(cls) -> "AppConfig":
        load_dotenv()
        return cls(
            openai_api_key=os.getenv("OPENAI_API_KEY", ""),
            openai_model=os.getenv("OPENAI_MODEL", "qwen-plus"),
            mcp_transport=os.getenv("MCP_TRANSPORT", "stdio"),
            ...
        )

2.2 客户端 client/*

client/runtime.py：按 MCP_TRANSPORT 选连接方式

协议方法三种一致，差别只在“怎么连”。stdio 走子进程；sse / http 走网络 URL。

def create_mcp_client(config: AppConfig) -> Client:
    if config.mcp_transport == "stdio":
        return create_stdio_client()                       # 子进程 + stdin/stdout
    if config.mcp_transport == "sse":
        return create_sse_client(config)                    # http://host:port/sse
    return create_streamable_http_client(config)           # http://host:port/mcp

client/llm.py：工具发现 + LLM 工具调用循环

这是教学价值最高的文件。逻辑分四步：(1) list_tools 发现工具 → (2) 转成 OpenAI tools → (3) 调 LLM 决策 → (4) LLM 要调就执行 MCP 工具并回填结果。

async def ask_with_llm(mcp_client, llm_client, config, user_prompt) -> str:
    openai_tools = to_openai_tools(await mcp_client.list_tools())
    messages = [
        {"role": "system", "content": "你是教学演示助手..."},
        {"role": "user", "content": user_prompt},
    ]
    for _ in range(config.llm_max_tool_rounds):
        completion = await asyncio.to_thread(
            llm_client.chat.completions.create,        # 阻塞 SDK -> 放进线程
            model=config.openai_model,
            messages=messages, tools=openai_tools,
            tool_choice="auto", temperature=0,
        )
        message = completion.choices[0].message
        tool_calls = message.tool_calls or []
        if not tool_calls:
            return (message.content or "").strip()      # 不调工具 = 最终答案
        messages.append(_assistant_message_with_calls(message))
        for call in tool_calls:
            args = json.loads(call.function.arguments or "{}")
            result = await mcp_client.call_tool(call.function.name, args)
            messages.append(_tool_result_message(call, result))
    return ""

其中两个小工具函数把对话历史拼装明了化（行为零变化，只是可读性）：

def _assistant_message_with_calls(message) -> dict:
    """把 LLM 的 tool 连同它的 tool_calls 一起打包成 assistant 消息。"""
    return {"role": "assistant", "content": message.content or "",
            "tool_calls": [{"id": c.id, "type": "function",
                            "function": {"name": c.function.name,
                                         "arguments": c.function.arguments}}
                           for c in (message.tool_calls or [])]}

def _tool_result_message(call, result) -> dict:
    """把一次工具调用结果回填成 role=tool 的消息，供 LLM 二次推理。"""
    return {"role": "tool", "tool_call_id": call.id,
            "name": call.function.name,
            "content": json.dumps({"result": _tool_result_text(result)},
                                  ensure_ascii=False)}

2.3 服务端 server/*

server/app.py：注册 4 个计算器工具

用 @mcp.tool 把普通 Python 函数登记成 MCP 工具。FastMCP 会读类型注解自动生成 inputSchema，所以完全不用手写 JSON Schema。

def create_mcp_server() -> FastMCP:
    mcp = FastMCP("Test Server")

    @mcp.tool(name="calculator_add", title="Calculator Add",
              description="Add two numbers")
    def add(a: int, b: int) -> int:
        return a + b
    # ... subtract / multiply / divide 同款登记
    # divide 里 b == 0 时抛 ValueError，让调用方看到非法状态
    return mcp

server/runtime.py：按协议启动

def run_server_by_transport(config: AppConfig) -> None:
    if config.mcp_transport == "stdio":           run_server_stdio(config)
    elif config.mcp_transport == "sse":           run_server_sse(config)
    else:                                         run_server_streamable_http(config)

2.4 主流程活动图

把“客户端”一整条主线画成活动图，包含三传输分支与 LLM↔MCP 循环。

3. 完整调用流程

以“帮我算 1+2”这一句话为线索，把整条调用走一遍。本次涉及 4 个角色：用户 / 客户端 / 服务端 / LLM。

3.1 时序图

分五个阶段：握手 → 发现 → 决策 → 执行 → 回填输出。下面是每段对应的真实 JSON-RPC 报文。

3.2 真实 JSON-RPC 报文（stdio 抓包）

① initialize — 握手请求

{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-11-25",
    "capabilities": {},
    "clientInfo": { "name": "mcp", "version": "0.1.0" }
  }
}

② initialize — 握手响应

{
  "jsonrpc": "2.0",
  "id": 0,
  "result": {
    "protocolVersion": "2025-11-25",
    "capabilities": {
      "prompts": { "listChanged": false },
      "resources": { "subscribe": false, "listChanged": false },
      "tools": { "listChanged": true }
    },
    "serverInfo": { "name": "Test Server", "version": "3.1.1" }
  }
}

③ notifications/initialized — 通知

{ "jsonrpc": "2.0", "method": "notifications/initialized" }

④ tools/list — 工具发现

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list"
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [{
      "name": "calculator_add",
      "title": "Calculator Add",
      "description": "Add two numbers",
      "inputSchema": {
        "type": "object",
        "properties": { "a": { "type": "integer" }, "b": { "type": "integer" } },
        "required": ["a", "b"]
      }
    }]
  }
}

⑤ tools/call — 执行工具

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "calculator_add",
    "arguments": { "a": 1, "b": 2 }
  }
}

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [{ "type": "text", "text": "3" }],
    "structuredContent": { "result": 3 },
    "isError": false
  }
}

4. 通用应用场景

本 demo 是“玩具版”，但 MCP 设计面向的是这些通用场景：

4.1 场景一：本地开发 / 演示

客户端直接把服务端当子进程拉起，用 stdio 通信。零部署、最方便调试。适合 IDE 插件、本地 Agent、教学。

4.2 场景二：远程 MCP 服务

工具逻辑跑在一个长期运行的进程（sse / streamable_http），多个客户端通过网络共享同一份工具实现。适合团队共用、跨机器、跨账号体系。

4.3 场景三：企业内部“工具总线”

把不同业务系统（数据库查询、工单、监控、CRM…）按 MCP 统一暴露成工具，聊天模型只对接 MCP 客户端，就能一站式调用全公司能力。

4.4 场景四：Agent 生态集成

第三方 Agent / IDE / 框架只要支持 MCP 客户端，就能“插上”你提供的工具，无需关心你的实现语言与运行环境。

4.5 MCP vs Function Calling

很多人以为两者是二选一，其实是分工：

• Function Calling：模型 API 内部能力，决定“要不要调、调哪个、传什么参数”。依赖具体厂商。
• MCP：模型与工具之间的通信协议层，标准化 tools/list / tools/call，传输可选 stdio/sse/http。

本 demo 的实际配合：LLM 用 Function Calling 做决策 → 客户端按 MCP 执行工具 → 回填结果。

5. 三种传输对比

MCP 协议本身在三种传输下完全一样，只有“字节怎么从客户端送到服务端”不同。

选型建议：

• 想最快跑通 → 用 stdio。
• 想模拟“客户端连远程服务” → 用 sse 或 streamable_http。
• 多人部署 / Web 化 → 优先 HTTP 形态（sse / streamable_http）。

5.1 怎么在本机跑

① stdio（最简单）

uv sync
uv run python main.py
# 输入：帮我算 1+2
# 输出：助手: 1 + 2 = 3

② sse（双终端）

# .env: MCP_TRANSPORT=sse

# 终端 A（服务端）
uv run python -m server.sse

# 终端 B（客户端）
uv run python main.py

③ streamable_http（双终端）

# .env: MCP_TRANSPORT=streamable_http

# 终端 A
uv run python -m server.streamable_http

# 终端 B
uv run python main.py