跳转至

第2章 MCP Server

学习目标

  • 掌握使用 mcp Python SDK 创建 Server 的完整流程
  • 能够定义 Tool / Resource / Prompt 三类原语
  • 理解 stdio 与 Streamable HTTP 两种传输模式的部署差异
  • 掌握 Server 的生命周期管理与错误处理
  • 能够为企业内部能力封装一个生产可用的 MCP Server

问题导入

第 1 章我们理解了 MCP 的协议价值，但"协议"只是契约，真正让能力落地的是 MCP Server。一个常见的困惑是：我已经有一个 REST API（比如内部订单查询接口），为什么还要再包一层 MCP Server？答案是——REST API 面向"程序调用者"，而 MCP Server 面向"模型调用者"。前者需要调用方知道 URL、方法、鉴权方式、参数结构；后者把这些细节全部封装在协议层，模型只需要通过标准化的 list_tools / call_tool 就能发现并使用。

换句话说，MCP Server 是把"现有能力"翻译成"模型友好的标准化能力"。本章我们将用 mcp Python SDK 从零构建一个功能完整的 Server，覆盖工具、资源、提示三类原语，并分别用 stdio 与 HTTP 两种方式部署。

核心概念

1. FastMCP：高层 Python SDK

mcp.server.fastmcp.FastMCP 是官方提供的高层封装，类似 FastAPI 之于 ASGI。它通过装饰器把普通 Python 函数自动转换为 MCP 原语，自动从类型注解生成 JSON Schema，极大降低开发成本。

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")

第一个参数是 Server 名称，会出现在 Client 端的能力发现结果中。

2. 三类原语的定义方式

Tool（工具）：用 @mcp.tool() 装饰函数，函数即工具，docstring 即描述，类型注解即 Schema。

@mcp.tool()
def search_order(order_id: str) -> dict:
    """根据订单号查询订单状态"""
    return {"order_id": order_id, "status": "shipped"}

Resource（资源）：用 @mcp.resource("uri") 装饰，URI 是资源标识，可带模板参数。

@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """读取本地文件内容"""
    return open(path, encoding="utf-8").read()

Prompt（提示模板）：用 @mcp.prompt() 装饰，返回一段预定义的提示文本。

@mcp.prompt()
def debug_error(error_msg: str) -> str:
    """生成排错提示"""
    return f"请分析以下错误的根因并给出修复建议：\n{error_msg}"

3. 异步与上下文

FastMCP 同时支持同步与异步函数。对于 IO 密集型操作（HTTP 调用、数据库查询），推荐使用 async def 以提升并发能力。Server 函数还可注入 Context 对象，用于日志、进度上报与读取请求方信息。

from mcp.server.fastmcp import Context

@mcp.tool()
async def long_task(n: int, ctx: Context) -> str:
    await ctx.report_progress(0, n)
    for i in range(n):
        await ctx.info(f"处理第 {i+1} 项")
        await ctx.report_progress(i+1, n)
    return "完成"

4. 传输模式与启动

FastMCP 通过 run() 方法启动，支持两种传输：

# stdio：作为子进程运行，适合本地工具
mcp.run(transport="stdio")

# Streamable HTTP：作为 HTTP 服务，适合远程部署
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

stdio 模式无需端口，由 Host 直接拉起；HTTP 模式则像一个普通 Web 服务，可被远程 Client 连接。

5. 生命周期与错误处理

Server 的生命周期由 run() 管理，启动后进入消息循环。工具函数中抛出的异常会被 SDK 自动捕获并转换为 MCP 错误响应返回给 Client，不会导致 Server 崩溃。但应注意：

  • 输入校验：用 Pydantic 模型作为参数类型，让 SDK 自动校验。
  • 超时控制：长任务应主动上报进度，避免 Client 超时。
  • 资源释放：在函数内用 with 语句管理文件、连接等资源。

原理图

flowchart TB
    subgraph Server["MCP Server 进程"]
        Entry["mcp.run(transport)"]
        Loop["JSON-RPC 消息循环"]
        Reg["原语注册表<br/>Tools / Resources / Prompts"]
        Handler["请求分发器"]
        T1["Tool: search_order"]
        T2["Tool: create_ticket"]
        R1["Resource: file://"]
        P1["Prompt: debug_error"]
    end

    Client["MCP Client"] <-.stdio / HTTP.-> Loop
    Entry --> Loop
    Loop --> Handler
    Handler --> Reg
    Reg --> T1
    Reg --> T2
    Reg --> R1
    Reg --> P1

    classDef server fill:#fff3e0,stroke:#f57c00
    classDef entry fill:#e3f2fd,stroke:#1976d2
    class Server,Loop,Reg,Handler,T1,T2,R1,P1 server
    class Entry,Client entry

图：MCP Server 内部结构——入口启动消息循环，请求经分发器路由到已注册的原语

示例代码

下面是一个相对完整的企业工具 MCP Server，覆盖工具、资源、提示三类原语，并使用 Pydantic 做参数校验。

# enterprise_server.py
# pip install "mcp[cli]" pydantic
from pathlib import Path
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("enterprise-tools")


# ---------- 1. 工具定义 ----------

class TicketInput(BaseModel):
    title: str = Field(description="工单标题")
    priority: str = Field(default="normal", description="优先级：low/normal/high/urgent")


@mcp.tool()
async def create_ticket(payload: TicketInput, ctx: Context) -> dict:
    """在工单系统创建一条新工单，返回工单号"""
    await ctx.info(f"正在创建工单：{payload.title}")
    # 模拟调用工单系统 API
    ticket_id = "TKT-" + str(hash(payload.title) % 10000)
    return {"ticket_id": ticket_id, "status": "created", "priority": payload.priority}


@mcp.tool()
async def query_metric(metric: str, window: int = 60) -> dict:
    """查询监控系统指标，window 为时间窗口（秒）"""
    metrics = {"cpu": 0.42, "memory": 0.68, "qps": 1234}
    return {"metric": metric, "value": metrics.get(metric, 0), "window_s": window}


# ---------- 2. 资源定义 ----------

@mcp.resource("doc://{name}")
def get_doc(name: str) -> str:
    """读取内部文档库中的文档"""
    docs = {
        "onboarding": "新员工入职指南：包含账号申请、权限申请流程……",
        "runbook": "故障应急 Runbook：先确认告警，再查日志，最后通知值班人……",
    }
    return docs.get(name, f"文档 {name} 不存在")


# ---------- 3. 提示模板 ----------

@mcp.prompt()
def incident_report(service: str, symptom: str) -> str:
    """生成故障复盘报告模板"""
    return (
        f"请根据以下信息撰写故障复盘报告：\n"
        f"- 受影响服务：{service}\n"
        f"- 故障现象：{symptom}\n"
        f"报告需包含：时间线、根因分析、影响范围、改进措施。"
    )


# ---------- 4. 启动 ----------
if __name__ == "__main__":
    import sys
    mode = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    if mode == "http":
        mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
    else:
        mcp.run(transport="stdio")

启动方式：

# stdio 模式（供本地 Host 拉起）
python enterprise_server.py stdio

# HTTP 模式（远程部署）
python enterprise_server.py http
# 之后可用 mcp dev 或 Client 连接 http://localhost:8000/mcp

企业案例

案例 1：企业工具服务化

某电商公司将"订单查询""库存调整""退款审批"等 8 个高频操作统一封装为一个 MCP Server，部署在内网。所有 Agent（客服、运营、风控）通过 HTTP 连接同一 Server，权限由 Server 端统一校验。新增工具只需在 Server 内加一个 @mcp.tool()，所有 Agent 立即可用。

案例 2：DevOps 工具网关

运维团队把 K8s 操作、日志查询、告警确认封装为 MCP Server，作为"DevOps 网关"。Cursor 中的运维 Agent 可直接调用这些工具完成故障排查，避免工程师在多个终端间切换。Server 内部用 RBAC 控制每个工具的调用权限。

案例 3：开源工具贡献

某团队开发了"企业 wiki 搜索"MCP Server，内部验证后开源到 GitHub。社区其他公司可以直接复用，团队也因此获得反馈改进了分页逻辑，形成正向循环。

常见问题

Q1：一个 Server 应该放多少个工具？ 建议按业务域组织，单个 Server 内工具数控制在 10 个以内。工具过多会让模型选择困难，且增加上下文负担。可按域拆分多个 Server。

Q2：HTTP 模式下如何做鉴权？ 在 Server 启动时配置中间件，从请求头读取 Token 并校验。FastMCP 支持自定义 ASGI 中间件。生产环境建议放在网关层统一鉴权。

Q3：工具抛异常会影响 Server 吗？ 不会。SDK 会捕获异常并返回错误响应给 Client，Server 进程继续运行。但应避免在工具内吞掉异常不报，否则 Client 无法感知失败。

Q4：stdio 模式下如何调试？ 使用官方 mcp dev CLI 工具，它会拉起 Server 并提供交互式界面查看工具列表、调用结果。也可在代码中写日志到文件。

Q5：Resource 和 Tool 的边界在哪？ Tool 是"模型主动调用、有副作用或需参数"的操作；Resource 是"应用读取、静态或半静态"的数据。读一个文件可作为 Resource，写文件则应为 Tool。

面试题

1. 简述用 FastMCP 创建一个工具的步骤。 答：① 实例化 FastMCP(name)；② 用 @mcp.tool() 装饰函数，docstring 作为描述，类型注解作为 Schema；③ 调用 mcp.run(transport=...) 启动。

2. stdio 与 Streamable HTTP 部署各自的优劣？ 答：stdio 无需端口、低延迟、适合本地工具，但不支持远程；HTTP 支持远程、可负载均衡、可加鉴权，但有网络开销，适合生产环境。

3. Context 对象有什么用？ 答：用于向 Client 上报进度（report_progress）、写日志（info/error）、读取请求元信息，让长任务可观测、可中断。

4. 如何让工具参数校验更健壮？ 答：用 Pydantic BaseModel 作为参数类型，SDK 会自动生成 Schema 并校验输入，避免手写校验逻辑。

5. 一个企业应该如何划分多个 MCP Server？ 答：按业务域或团队边界划分，例如订单 Server、运维 Server、知识库 Server。每个 Server 聚焦单一职责，便于独立部署与权限管控。

练习题

  1. 扩展示例 Server，新增一个 list_orders 工具，支持按状态过滤并分页返回，用 Pydantic 定义分页参数。
  2. 把示例改为 HTTP 模式启动，并用 mcp dev http://localhost:8000/mcp 验证工具可被发现与调用。
  3. 为 create_ticket 工具添加进度上报：模拟"校验→提交→通知"三个阶段，每阶段用 ctx.report_progress 上报。

小结

本章用 mcp Python SDK 完整实现了一个企业级 MCP Server，覆盖工具、资源、提示三类原语，并支持 stdio 与 HTTP 两种部署方式。核心要点：用 FastMCP 装饰器降低开发成本；用 Pydantic 做参数校验；按业务域拆分 Server；HTTP 模式适合生产远程部署。下一章我们将转向 Client 侧，看 Agent 如何发现并调用这些 Server 暴露的能力。