MCP 协议开发实战

从零构建你的第一个 MCP Server，让 AI 成为万能工具人。

---

附录

A. MCP API 速查表（Python SDK）

FastMCP 初始化

from mcp.server.fastmcp import FastMCP

# 创建 Server 实例
mcp = FastMCP(
    "Server 名称",          # name：Server 名称
    instructions="...",     # 可选：给 AI 的使用说明
)

Tool 注册

# ═══════════════════════════════════════════
# 基础 Tool
# ═══════════════════════════════════════════

@mcp.tool()
def my_tool(param1: str, param2: int = 10) -> str:
    """工具描述（会被 AI 看到）。

    Args:
        param1: 参数1的说明
        param2: 参数2的说明，默认 10
    """
    return f"结果：{param1}, {param2}"

# ═══════════════════════════════════════════
# 异步 Tool
# ═══════════════════════════════════════════

@mcp.tool()
async def async_tool(url: str) -> str:
    """异步工具"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

# ═══════════════════════════════════════════
# 带 Context 的 Tool
# ═══════════════════════════════════════════

from mcp.server.fastmcp import Context

@mcp.tool()
async def tool_with_ctx(query: str, ctx: Context) -> str:
    """带日志和进度的工具"""
    ctx.info("开始处理")                          # 日志
    await ctx.report_progress(1, 3, "步骤1")     # 进度
    # ... 处理逻辑
    return "结果"

# ═══════════════════════════════════════════
# 自定义名称和描述
# ═══════════════════════════════════════════

@mcp.tool(name="custom_name", description="自定义描述")
def renamed_tool(x: int) -> str:
    return str(x)

参数类型映射

Python 类型注解 → JSON Schema 映射：

  str                  → {"type": "string"}
  int                  → {"type": "integer"}
  float                → {"type": "number"}
  bool                 → {"type": "boolean"}
  list[str]            → {"type": "array", "items": {"type": "string"}}
  dict[str, Any]       → {"type": "object"}
  str | None           → {"type": "string"} (nullable)
  Literal["a", "b"]    → {"type": "string", "enum": ["a", "b"]}

  Pydantic BaseModel   → 嵌套对象 Schema
  Field(description=)  → 字段描述
  Field(ge=0, le=100)  → 数值范围约束

Resource 注册

# ═══════════════════════════════════════════
# 静态 Resource（固定 URI）
# ═══════════════════════════════════════════

@mcp.resource("config://app/settings")
def app_settings() -> str:
    """应用配置信息"""
    return json.dumps({"version": "1.0", "debug": False})

# ═══════════════════════════════════════════
# Resource 模板（参数化 URI）
# ═══════════════════════════════════════════

@mcp.resource("file://docs/{filename}")
def read_doc(filename: str) -> str:
    """读取文档文件"""
    return Path(f"docs/{filename}").read_text()

# ═══════════════════════════════════════════
# 指定 MIME 类型
# ═══════════════════════════════════════════

@mcp.resource("data://report.json", mime_type="application/json")
def report_data() -> str:
    return json.dumps({"total": 42})

Prompt 注册

# ═══════════════════════════════════════════
# 基础 Prompt
# ═══════════════════════════════════════════

@mcp.prompt()
def review(code: str, language: str = "python") -> str:
    """代码审查模板"""
    return f"请审查以下 {language} 代码：\n\n```{language}\n{code}\n```"

# ═══════════════════════════════════════════
# 多轮消息 Prompt
# ═══════════════════════════════════════════

from mcp.types import UserMessage, AssistantMessage, TextContent

@mcp.prompt()
def debug_session(error: str) -> list:
    """调试会话模板"""
    return [
        UserMessage(content=TextContent(type="text", text=f"我遇到了报错：{error}")),
        AssistantMessage(content=TextContent(type="text", text="我来帮你分析。请提供以下信息：...")),
        UserMessage(content=TextContent(type="text", text="请根据上述信息给出解决方案。")),
    ]

Server 启动

# STDIO 模式（本地，默认）
mcp.run()
mcp.run(transport="stdio")

# SSE 模式（远程）
mcp.run(transport="sse")

# Streamable HTTP 模式（远程，推荐）
mcp.run(transport="streamable-http")

Client 连接

from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp.client.sse import sse_client

# ═══════════════════════════════════════════
# STDIO Client
# ═══════════════════════════════════════════

server_params = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
)
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()
        result = await session.call_tool("tool_name", {"param": "value"})

# ═══════════════════════════════════════════
# SSE Client
# ═══════════════════════════════════════════

async with sse_client("http://server:8000/sse") as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        resources = await session.list_resources()
        content = await session.read_resource("config://app/settings")

---

B. MCP API 速查表（TypeScript SDK）

McpServer 初始化

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "Server 名称",
  version: "1.0.0",
});

Tool 注册

// ═══════════════════════════════════════════
// 基础 Tool
// ═══════════════════════════════════════════

server.tool(
  "tool_name",                           // 工具名称
  "工具描述",                              // 描述（AI 可见）
  {                                       // 参数 Schema（Zod）
    param1: z.string().describe("参数1说明"),
    param2: z.number().int().default(10).describe("参数2说明"),
  },
  async ({ param1, param2 }) => ({       // 处理函数
    content: [{
      type: "text" as const,
      text: `结果：${param1}, ${param2}`,
    }],
  })
);

// ═══════════════════════════════════════════
// 返回错误
// ═══════════════════════════════════════════

server.tool("risky_tool", "可能失败的工具", {
  input: z.string(),
}, async ({ input }) => ({
  content: [{ type: "text" as const, text: "出错了" }],
  isError: true,   // ← 标记为错误
}));

Zod 类型映射

Zod Schema → JSON Schema 映射：

  z.string()              → {"type": "string"}
  z.number().int()        → {"type": "integer"}
  z.number()              → {"type": "number"}
  z.boolean()             → {"type": "boolean"}
  z.array(z.string())     → {"type": "array", ...}
  z.object({...})         → {"type": "object", ...}
  z.enum(["a", "b"])      → {"type": "string", "enum": ["a","b"]}
  .optional()             → 可选参数
  .default(10)            → 默认值
  .describe("说明")       → description 字段

Resource 注册

// ═══════════════════════════════════════════
// 静态 Resource
// ═══════════════════════════════════════════

server.resource(
  "app-config",                          // 资源名称
  "config://app/settings",               // URI
  async () => ({
    contents: [{
      uri: "config://app/settings",
      mimeType: "application/json",
      text: JSON.stringify({ version: "1.0" }),
    }],
  })
);

// ═══════════════════════════════════════════
// Resource 模板
// ═══════════════════════════════════════════

server.resource(
  "doc-file",
  "file://docs/{filename}",              // URI 模板
  async ({ filename }) => ({
    contents: [{
      uri: `file://docs/${filename}`,
      mimeType: "text/plain",
      text: await readFile(`docs/${filename}`, "utf-8"),
    }],
  })
);

Prompt 注册

server.prompt(
  "code_review",                         // Prompt 名称
  "代码审查模板",                          // 描述
  {                                       // 参数（Zod）
    code: z.string().describe("要审查的代码"),
    language: z.string().default("python"),
  },
  async ({ code, language }) => ({
    messages: [{
      role: "user" as const,
      content: {
        type: "text" as const,
        text: `请审查以下 ${language} 代码：\n\n\`\`\`${language}\n${code}\n\`\`\``,
      },
    }],
  })
);

Server 启动

// STDIO 模式
const transport = new StdioServerTransport();
await server.connect(transport);

// SSE 模式（需要额外设置 HTTP 服务器）
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
// ... 配合 Express / Fastify 使用

Python vs TypeScript API 对照表

功能	Python (FastMCP)	TypeScript (McpServer)
注册 Tool	@mcp.tool()	server.tool(name, desc, schema, fn)
注册 Resource	@mcp.resource(uri)	server.resource(name, uri, fn)
注册 Prompt	@mcp.prompt()	server.prompt(name, desc, schema, fn)
参数校验	类型注解 + Pydantic	Zod Schema
返回值	直接 return "字符串"	return { content: [...] }
错误标记	raise ToolError(...)	return { ..., isError: true }
启动	mcp.run()	server.connect(transport)
包管理	uv add "mcp[cli]"	npm install @modelcontextprotocol/sdk

---

C. 常见报错与解决方案

开发 MCP Server 时你大概率会遇到以下这些报错。按出现频率排序：

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 1：ModuleNotFoundError: No module named 'mcp'
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：没有安装 MCP SDK，或安装在了错误的虚拟环境中

  解决：
    # 确保用 uv 安装
    uv add "mcp[cli]"

    # 确认安装成功
    uv run python -c "import mcp; print(mcp.__version__)"

    # 如果用 pip
    pip install "mcp[cli]"

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 2：Server 启动后 Claude Desktop 中看不到工具
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：claude_desktop_config.json 配置错误，或路径不对

  排查步骤：
    1. 确认配置文件路径正确：
       macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
       Windows: %APPDATA%\Claude\claude_desktop_config.json

    2. 检查 JSON 格式（逗号、引号）：
       $ python -m json.tool < claude_desktop_config.json

    3. 确认 command 和 args 能在终端中正常运行：
       $ uv run /你的路径/server.py
       → 应该没有报错，等待输入

    4. 重启 Claude Desktop（修改配置后必须重启）

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 3：TypeError: object str can't be used in 'await' expression
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：对非异步函数使用了 await

  错误代码：
    @mcp.tool()
    async def my_tool() -> str:
        result = await sync_function()  # ← sync_function 不是 async！
        return result

  解决：
    # 方案 1：去掉 await（如果函数是同步的）
    result = sync_function()

    # 方案 2：改函数为 async
    async def sync_function():
        ...

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 4：ValidationError - Input should be a valid string
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：参数类型不匹配（Pydantic 校验失败）

  场景：
    @mcp.tool()
    def my_tool(count: int) -> str:  # 期望 int
        ...

    # 但 AI 传了 "abc"（字符串），Pydantic 无法转换

  解决：
    → Tool 的 docstring 中明确说明参数格式
    → 使用 Annotated + Field 添加约束
    → 在函数内部做额外校验

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 5：httpx.ConnectError: Connection refused
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：SSE Server 没有启动，或端口/地址不对

  排查：
    # 确认 Server 在运行
    $ curl http://localhost:8000/sse
    → 应该返回 SSE 事件流

    # 如果是远程 Server
    $ curl http://你的IP:8000/sse
    → 检查防火墙、安全组

    # 常见坑：Server 只监听 127.0.0.1
    → 改为 HOST=0.0.0.0 uv run server.py

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 6：RuntimeWarning: coroutine 'xxx' was never awaited
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：调用了 async 函数但忘了 await

  错误代码：
    async def fetch_data():
        return "data"

    @mcp.tool()
    async def my_tool() -> str:
        result = fetch_data()  # ← 缺少 await！返回的是协程对象
        return str(result)     # → 输出 <coroutine object ...>

  解决：
    result = await fetch_data()  # ← 加上 await

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 7：mcp.shared.exceptions.McpError: Method not found
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：Client 尝试调用了 Server 不支持的方法

  场景：
    → Client 调用 tools/call，但 Server 没注册任何 Tool
    → Client 调用的 Tool 名称拼写错误

  解决：
    1. 先调用 tools/list 确认可用工具列表
    2. 检查 Tool 名称是否拼写正确
    3. 用 MCP Inspector 验证 Server 注册了哪些功能

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 8：json.decoder.JSONDecodeError
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：Server 代码中有 print() 语句污染了 STDIO 通道

  场景：
    @mcp.tool()
    def my_tool() -> str:
        print("debug info")  # ← 这会破坏 JSON-RPC 通信！
        return "result"

  解决：
    → 删除所有 print() 语句
    → 改用 logging 或 ctx.info() 输出调试信息
    → 或改用 sys.stderr 输出

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 9：TS SDK - Cannot find module '@modelcontextprotocol/sdk'
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：npm 包没有安装，或 tsconfig.json 配置不对

  解决：
    # 重新安装
    npm install @modelcontextprotocol/sdk

    # 检查 tsconfig.json
    "module": "Node16",
    "moduleResolution": "Node16"

    # 检查 package.json
    "type": "module"

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
❌ 报错 10：Tool 返回内容被截断
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  原因：返回内容太长，超出 Client 或 LLM 的处理能力

  解决：
    @mcp.tool()
    def read_file(path: str) -> str:
        content = Path(path).read_text()
        # 限制返回长度
        if len(content) > 10000:
            content = content[:10000] + "\n\n... 内容已截断"
        return content

  建议：
    → Tool 返回内容控制在 10KB 以内
    → 大数据集做分页处理
    → 提供摘要而非全量数据

D. 推荐学习资源与社区

官方资源（首选）

MCP 官方资源清单：

  📖 MCP 规范文档（权威参考）
     https://spec.modelcontextprotocol.io
     → 协议的完整技术规范
     → 包括 JSON-RPC 消息格式、生命周期、错误码
     → 适合深入理解协议细节

  🏠 MCP 官方网站
     https://modelcontextprotocol.io
     → 入门教程、概念介绍
     → 架构图、快速上手指南
     → 适合初学者

  🐍 Python SDK（GitHub）
     https://github.com/modelcontextprotocol/python-sdk
     → FastMCP 高层 API 源码
     → 丰富的 examples/ 目录
     → Issues 区有很多实际问题的讨论

  📦 TypeScript SDK（GitHub）
     https://github.com/modelcontextprotocol/typescript-sdk
     → McpServer API 源码
     → 配套 Zod 类型定义
     → 参考 examples/ 目录

  🔍 MCP Inspector
     https://github.com/modelcontextprotocol/inspector
     → 在线调试工具
     → 开发 MCP Server 的必备工具

社区 MCP Server 合集

优质 MCP Server 仓库（学习参考）：

  🌟 awesome-mcp-servers
     https://github.com/punkpeye/awesome-mcp-servers
     → 社区维护的 MCP Server 合集
     → 按功能分类：文件系统、数据库、API、开发工具
     → 寻找灵感和参考实现的好地方

  🌟 Anthropic 官方 Servers
     https://github.com/modelcontextprotocol/servers
     → 官方维护的参考 Server 实现
     → 包括：文件系统、GitHub、Slack、PostgreSQL 等
     → 代码质量高，适合作为模板

  热门社区 Server 举例：
     • mcp-server-sqlite   → SQLite 数据库操作
     • mcp-server-github   → GitHub API 集成
     • mcp-server-slack     → Slack 消息管理
     • mcp-server-filesystem → 本地文件系统访问
     • mcp-server-puppeteer → 浏览器自动化

学习路径建议

MCP 学习路线图：

  阶段 1：入门（1-2 天）
  ────────────────────
    ✅ 读完本教程第 1-4 章
    ✅ 用 FastMCP 创建第一个 Server
    ✅ 用 MCP Inspector 调试
    ✅ 连接 Claude Desktop 实际使用
    → 目标：能跑起来，理解基本概念

  阶段 2：深入（3-5 天）
  ────────────────────
    ✅ 读完本教程第 5-7 章
    ✅ 掌握 Tools / Resources / Prompts 的进阶用法
    ✅ 理解异步处理、错误处理、参数校验
    ✅ 构建一个实际有用的 Server
    → 目标：能开发生产可用的 MCP Server

  阶段 3：生产化（1 周）
  ────────────────────
    ✅ 读完本教程第 8-10 章
    ✅ 掌握 Docker 部署和远程连接
    ✅ 实现认证授权和安全防护
    ✅ 配置日志监控
    → 目标：能部署和运维 MCP Server

  阶段 4：进阶（持续）
  ────────────────────
    □ 阅读 MCP 协议规范原文
    □ 研究社区 Server 的实现
    □ 为开源项目贡献 MCP Server
    □ 探索 MCP + Agent 的组合模式
    → 目标：成为 MCP 生态贡献者

进阶方向

学完本教程后，可以深入的方向：

  1. MCP + AI Agent
     → 将 MCP Server 作为 Agent 的工具集
     → 结合 LangChain / LangGraph 构建智能体
     → 让 Agent 自主决定调用哪些 MCP 工具

  2. 多 Server 编排
     → 一个 Host 连接多个 MCP Server
     → 跨 Server 的工作流编排
     → 微服务化：每个 Server 负责一个领域

  3. MCP Client 开发
     → 不依赖 Claude Desktop，自己实现 Host
     → 集成到自己的应用中
     → 构建自定义的 AI 助手平台

  4. 协议扩展与贡献
     → 参与 MCP 规范的讨论
     → 为 SDK 提交 PR
     → 发布自己的 MCP Server 到社区

相关技术栈推荐

方向	技术	说明
AI 框架	LangChain / LangGraph	构建 AI Agent，可对接 MCP
HTTP 框架	FastAPI / Starlette	MCP SSE Transport 底层依赖
参数校验	Pydantic / Zod	MCP 两大 SDK 的参数校验基础
异步编程	asyncio / aiohttp	Python 异步生态，MCP 大量使用
容器化	Docker / Kubernetes	生产部署基础设施
监控	Prometheus / Grafana	生产级指标收集和可视化

---

🎉 恭喜你读完了整个教程！ 从"MCP 是什么"到"生产部署"，你已经掌握了构建 MCP Server 的全部核心知识。现在，去构建你自己的 MCP Server，让 AI 成为真正的万能工具人吧！