使用 Python MCP SDK 构建生产级 MCP 服务器:FastMCP 专家实践指南

原创2026-09-08 14:30:361,957 阅读
文章标签:文档知识库AI 应用

使用 Python MCP SDK 构建生产级 MCP 服务器:FastMCP 专家实践指南

在 GitHub Copilot 生态中,通过 Agent 与 Skill 扩展能力时,MCP(Model Context Protocol)服务器是连接 LLM 与外部数据、工具和 API 的标准桥梁。本指南以 awesome-copilot 仓库中收录的 python-mcp-expert Agent 定义为骨架,系统讲解如何使用官方 Python MCP SDK 的 FastMCP 高层封装开发类型安全、健壮、易于被 LLM 理解与调用的生产级 MCP 服务器。你将掌握 stdio 与 streamable-http 双传输配置、Tool / Resource / Prompt 三大注册机制、结构化输出、上下文(Context)编程、lifespan 资源管理等完整实战方案,并能直接通过 uv run mcp dev 与 uv run mcp install 进行测试与集成。

一、整体定位:从 Agent 文档看 Python MCP 服务端的核心能力

python-mcp-expert.agent.md 定义了"世界级 MCP 服务器构建专家"应具备的知识结构,它可被概括为六条主线:

  1. 语言与 SDK 基础:精通 mcp 包、FastMCP、底层 Server 类、全部传输层与工具函数,面向 Python 3.10+ 的类型注解、async/await、装饰器与上下文管理器;
  2. 协议理解:完整掌握 Model Context Protocol 规范与能力协商;
  3. 数据传输:精通 stdio 与 streamable HTTP 两种传输方式,含 ASGI 挂载;
  4. 工具设计:编写类型安全、schema 正确的工具,并产出结构化结果;
  5. 最佳实践:测试、错误处理、日志、资源管理与安全;
  6. 调试能力:排查类型注解、schema 与传输层问题。

仓库中配套的 python-mcp-server.instructions.md 以 applyTo: '**/*.py, **/pyproject.toml, **/requirements.txt' 声明了适用范围,意味着该套方法论面向任何 Python MCP 服务端文件生效;python-mcp-server-generator Skill 则进一步把上述能力固化为"一键生成完整可运行 MCP 项目"的实操规范。三份文档共同构成"知识 → 规则 → 生成"的完整链路,本文即沿此链路展开。

二、工程初始化:用 uv 搭建 MCP 服务器项目

Agent 与 Skill 一致推荐使用 uv 作为项目与依赖管理工具。整个项目初始化只需两条命令:

uv init mcp-server-demo
uv add "mcp[cli]"
  • uv init 生成标准 Python 项目骨架(含 pyproject.toml);
  • uv add "mcp[cli]" 安装官方 Python MCP SDK,其中 [cli] extra 提供 mcp 命令行工具,这是后续 mcp dev(Inspector 调试)与 mcp install(安装到 Claude Desktop)的前提。

随后创建主服务文件(如 server.py),并保证其支持直接执行(if __name__ == "__main__" 保护),这是后续"一行命令运行、一键安装"的基础。

FastMCP 的导入路径为 mcp.server.fastmcp,服务器通过一次初始化即可定义全部能力:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My Server")   # 服务名称,可附加 instructions 等元信息

三、三大注册机制:Tool、Resource、Prompt

MCP 服务器对外暴露的三类能力分别通过三个装饰器注册,这是 Agent 文档强调的 Decorator Pattern:

  • @mcp.tool():注册可被 LLM 调用的工具函数,参数与返回值的类型注解自动生成 JSON Schema;
  • @mcp.resource():注册静态或动态资源,供客户端按 URI 读取(数据读取而非执行);
  • @mcp.prompt():注册可复用的提示词模板,返回结构化 Message 列表。

3.1 Tool:类型注解驱动 schema 生成

最简单的 stdio 服务器示例(来自 python-mcp-server.instructions.md):

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My Server")

@mcp.tool()
def calculate(a: int, b: int, op: str) -> int:
    """Perform calculation"""
    if op == "add":
        return a + b
    return a - b

if __name__ == "__main__":
    mcp.run()  # stdio by default

关键在于:函数签名中的完整类型注解不是可选项。Python MCP SDK 通过 inspect.signature 解析工具函数的参数与返回类型注解,结合 Pydantic 动态生成 JSON Schema,用于参数校验与客户端工具描述。文档串(docstring)会被写入工具描述(description)字段,成为 LLM 判断"何时调用该工具"的核心依据——因此 Agent 指南强调"清晰的 docstring 就是协议中的工具描述"。

3.2 结构化输出:让结果可被机器读取

当工具的返回类型是可被 Pydantic 识别的模型(BaseModel、TypedDict、dataclass)时,SDK 会自动将返回值转换为结构化内容(structured content),同时为兼容旧客户端保留文本内容。示例:

from pydantic import BaseModel, Field

class WeatherData(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    condition: str
    humidity: float

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """Get weather for a city"""
    return WeatherData(temperature=22.5, condition="sunny", humidity=65.0)

值得注意的细节:Field(description=...) 的描述会进入生成的 JSON Schema,为每个参数提供语义约束,因此 Agent 指南要求"用带描述的 Pydantic Field 校验输入、提供有意义的参数名与描述"。结构化输出的核心价值是让结果不仅是给 LLM 阅读的文本,更是可供上层程序直接消费的数据。

3.3 Resource:静态资源与 URI 模板动态资源

资源用 URI 标识,动态资源通过 URI 模板(URI template)暴露参数化数据:

@mcp.resource("users://{user_id}")
def get_user(user_id: str) -> str:
    """Get user profile data"""
    return f"User {user_id} profile data"

SDK 会自动从 users://{user_id} 这类模板 URI 中提取路径参数 user_id 并注入函数。模板中的参数名必须与函数参数名对应。文档中"以 resource://{param} 定义动态资源"即指此类模式。

3.4 Prompt:结构化 Message 序列

提示词函数返回 list[base.Message] 而非拼接字符串,确保消息结构(系统/用户/助手角色)明确:

from mcp.server.fastmcp.prompts import base

@mcp.prompt(title="Code Review")
def review_code(code: str) -> list[base.Message]:
    """Create code review prompt"""
    return [
        base.UserMessage("Review this code:"),
        base.UserMessage(code),
        base.AssistantMessage("I'll review the code for you.")
    ]

四、传输层选择:stdio(本地)与 streamable-http(远程)

Agent 的 Approach 明确要求"先弄清用例:服务器用于本地(stdio)还是远程(HTTP)"。这是决定服务器架构的首要问题。

4.1 stdio:本地默认传输

mcp.run() 不带参数即默认 stdio 传输,子进程通过标准输入/输出与宿主(如 Claude Desktop、VS Code)通信。文档明确指出一个关键约束:stdio 场景下日志必须写到 stderr,避免污染 stdout 上的协议消息流。

4.2 streamable-http:远程服务

mcp = FastMCP("My HTTP Server")

@mcp.tool()
def hello(name: str = "World") -> str:
    """Greet someone"""
    return f"Hello, {name}!"

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

HTTP 服务器支持按需启动与横向扩展,主要高级配置项包括:

配置/能力 说明 典型值/场景
stateless_http=True 启用无状态 HTTP 模式 多副本负载均衡、Serverless 场景
json_response=True 兼容现代客户端,启用 JSON 响应封装 面向支持 JSON 响应的客户端
CORS 配置 供浏览器客户端跨域访问 需暴露 Mcp-Session-Id 响应头
host / port 服务监听地址与端口 通过环境变量注入,如 host="0.0.0.0"

其中无状态模式需要说明:默认 Streamable HTTP 服务器是有状态的,靠 Mcp-Session-Id 头维持会话连续性(同一会话内可直接发送通知、复用上下文);启用 stateless_http=True 后每次请求独立、无持久会话,从而可在无共享内存的集群中扩展。从 mcp-implementation-security-review Skill 的传输分类表可以印证:host="0.0.0.0" 属于全网卡暴露(Network-exposed),而 host="127.0.0.1" 仅本地可达——远程部署时必须据此评估安全边界。

4.3 无状态会话与安全边界的关联

仓库中的 mcp-implementation-security-review Skill 为远程服务器建立了 MCP-01~MCP-05 基线控制,其中 MCP-02(会话)明确指出:Streamable HTTP 的 Mcp-Session-Id 只是关联/连续性令牌(correlation token),绝不能充当身份凭证——会话 ID 不具备权限授予语义,认证与授权必须作用于每一个请求。若在代码中直接信任会话 ID 作为 bearer 凭证,即构成混淆代理(confused deputy)风险。这条原则对任何落地 streamable-http 的实现都成立。

五、Context 编程:日志、进度、采样与用户交互

当工具需要日志、进度上报或与 LLM 交互时,可在工具/资源函数中注入 Context 参数。指令文档与 Agent 文档给出了完整 API 面:

from mcp.server.fastmcp import Context
from mcp.server.session import ServerSession

@mcp.tool()
async def process_data(data: str, ctx: Context[ServerSession, None]) -> str:
    """Process data with logging"""
    await ctx.info(f"Processing: {data}")
    await ctx.report_progress(0.5, 1.0, "Halfway done")
    return f"Processed: {data}"

Context 提供四类核心能力:

  1. 分级日志:await ctx.debug()、await ctx.info()、await ctx.warning()、await ctx.error()。不同于 print 到 stdout,这些调用通过 MCP 协议通道发往客户端,不会破坏 stdio 消息流;
  2. 进度上报:await ctx.report_progress(progress, total, message),三个参数依次为当前进度、总进度、进度说明文字;
  3. LLM 采样(sampling):工具内部请求宿主模型再生成文本,实现"LLM 驱动型工具"。底层通过 ctx.session.create_message(messages, max_tokens) 发起采样请求,message 由 SamplingMessage 与 TextContent(type="text", text=...) 构造:
from mcp.server.fastmcp import Context
from mcp.types import SamplingMessage, TextContent

@mcp.tool()
async def summarize(text: str, ctx: Context) -> str:
    """Summarize text using LLM"""
    result = await ctx.session.create_message(
        messages=[SamplingMessage(
            role="user",
            content=TextContent(type="text", text=f"Summarize: {text}")
        )],
        max_tokens=100
    )
    return result.content.text if result.content.type == "text" else ""
  1. 用户澄清(elicitation):await ctx.elicit(message, schema) 可在运行时请求宿主向用户追问输入,配合 schema 约束期望答复结构,适用于参数不完整或需要人工确认的交互式流程。

六、Lifespan:共享资源的启动与销毁

数据库连接池、HTTP 客户端等生命周期资源应当通过 lifespan 上下文管理器管理,这比在单个工具内反复建立/释放连接更高效且更安全:

from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import FastMCP, Context

@dataclass
class AppContext:
    db: Database

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()

mcp = FastMCP("My App", lifespan=app_lifespan)

@mcp.tool()
def query(sql: str, ctx: Context) -> str:
    """Query database"""
    db = ctx.request_context.lifespan_context.db
    return db.execute(sql)

该模式有两个关键点:

  • lifespan 函数通过 yield 向 FastMCP 暴露初始化好的上下文对象;finally 块保证服务器关闭时资源被回收;
  • 工具内通过 ctx.request_context.lifespan_context 访问该共享对象(指令文档明确给出该访问路径),从而在任意工具中安全复用连接。

七、错误处理与代码规范

7.1 错误处理范式

Agent 指南要求"实现完整的 try-except 并给出清晰错误信息"。FastMCP 工具内推荐模式:

@mcp.tool()
async def risky_operation(input: str) -> str:
    """Operation that might fail"""
    try:
        result = await perform_operation(input)
        return f"Success: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

此外还应遵守:I/O 密集操作一律使用 async 函数;资源清理放在 finally 块或上下文管理器中;输入提前校验(配合 Pydantic Field description)以减少下游异常。

7.2 编码与安全规范

综合 Agent、指令与 SKILL.md 三份来源,落地到代码时应坚持:

  • 全函数使用完整类型注解(驱动 schema 生成与校验);
  • 工具函数职责单一、参数名与描述有实际含义;
  • 用环境变量承载配置,日志走 stderr 或 Context 通道;
  • 当工具暴露文件系统或网络访问权限时,必须做安全评估(如路径白名单、URL 校验),避免把任意用户输入直接拼入 shell 或模板——这对应 mcp-implementation-security-review 中列出的命令注入、路径穿越、SSTI、SSRF 等 RCE 向量,是任何暴露给 LLM 的工具都绕不开的攻击面。

八、测试与集成:mcp dev 与 mcp install

Agent 指南将"尽早测试"列为工作方式的一部分,测试与部署命令集中在 mcp[cli] 附带的命令行工具上:

# 启动 MCP Inspector(交互式调试面板,可逐个调用工具、查看 schema 与协议消息)
uv run mcp dev server.py

# 安装到 Claude Desktop 客户端
uv run mcp install server.py

运行方式本身则取决于传输类型:

  • stdio 服务器:python server.py 或 uv run server.py,随后在客户端侧完成进程式接入;
  • HTTP 服务器:启动后客户端连接 http://localhost:PORT/mcp(Streamable HTTP 端点默认挂在 /mcp 路径)。

从 SKILL.md 的测试建议看,最佳实践是先独立验证每个工具(含错误输入、边界参数)再接入 LLM,即在"工具的确定性测试"与"LLM 自由调用"之间划清界限。

九、高级能力:图标、图像、补全与多服务器挂载

Agent 文档在"Advanced Capabilities"一节列出了面向生产体验的高级特性,在此统一说明其用法与适用场景:

能力 API / 用法 适用场景
图标配置 Icon(src="path", mimeType="image/png"),可应用于服务器、工具、资源与提示词 在 UI 型客户端中展示可视化标识
图像处理 return Image(data=bytes, format="png") 工具产出图片(截图、图表等)时自动封装为图像内容
参数补全 在工具参数处理器中接收部分值并返回候选建议 提升"参数含枚举/ID 列表"工具的输入体验
ASGI 挂载 mcp.streamable_http_app(),配合 Starlette/FastAPI 的 Mount("/path", ...) 复杂部署场景:鉴权中间件、路由整合
多服务器挂载 在同一 ASGI 应用中对不同路径 Mount 多个 FastMCP 实例 一组后端服务聚合暴露为多个 MCP 端点
低层 API 直接使用 Server 类 FastMCP 无法满足的精细控制需求(如游标分页)

其中 ASGI 挂载需要专门补充说明:FastMCP 通过 mcp.streamable_http_app() 暴露一个符合 Streamable HTTP 协议的 ASGI 应用,可直接嵌入既有 Starlette/FastAPI 服务,且一个 ASGI 应用可 Mount 多个 FastMCP 实例、各占不同路径。浏览器客户端跨域访问时,需要配置 CORS 并暴露 Mcp-Session-Id 头,否则有状态会话无法维持。

十、安全与生产化:从可用到可信

Agent 指南将测试、错误处理、日志、资源管理、安全并列为"最佳实践"五要素。结合仓库中 mcp.json 展现的本地 MCP 客户端形态(type: "local"、command: gh),可以得出一个结论:本仓库自身大量使用的是本地 stdio 子进程式 MCP,其安全边界与远程 HTTP 服务器截然不同。

对远程服务器的安全加固,应参考 mcp-implementation-security-review Skill 的五项基线控制自查:

  1. MCP-01 身份隔离:每个入站请求都经过可信身份提供方认证,在服务器边界执行授权,绝不用会话 ID、历史请求或网络位置推断身份;
  2. MCP-02 会话安全:会话 ID 仅作关联令牌,需 CSPRNG 生成、不可预测,且永不内嵌 URL、永不被当作凭据;
  3. MCP-03 限流:在 MCP 服务器运行时(而非仅在网关)按身份与会话双重维度限流,写操作与高成本工具采用更严格配额,超限关闭式失败(HTTP 429 + Retry-After);
  4. MCP-04 Schema 校验:所有工具参数在服务端按显式 schema 在执行前完成校验,默认拒绝未声明属性(additionalProperties: false 语义);
  5. MCP-05 SDK-first:优先使用官方 MCP SDK(Python 属 Tier 1 完全支持),避免手写协议栈导致的原语缺失。

工具设计上应主动规避 RCE 向量(命令注入、动态代码求值、不安全反序列化、路径穿越、SSTI、依赖劫持、SSRF),并对工具输出做数据最小化——这既符合 MCP-04 与 MCP06/MCP10(OWASP MCP Top 10)要求,也直接呼应了 Agent 文档"安全是生产级服务器的组成部分"的立场。

十一、小结:一套可落地的 Python MCP 服务器开发方法论

将 python-mcp-expert.agent.md 的能力清单落到工程实践中,可提炼为以下可执行路径:

  1. 选型先行:判断本地(stdio)还是远程(HTTP),决定传输、会话与安全模型;
  2. FastMCP 默认、低层 API 兜底:绝大多数场景使用 @mcp.tool()/@mcp.resource()/@mcp.prompt() 三大装饰器即可,只有需要精细控制(如游标分页)时才下沉到 Server 类;
  3. 类型注解即 schema:参数、返回值全量注解,docstring 精炼准确,Pydantic Field(description=...) 补足语义,结构化输出用 BaseModel/TypedDict;
  4. Context 贯穿横切关注点:日志、进度、采样、澄清全部走 Context,不污染传输流;
  5. lifespan 管好长生命周期资源,finally 保证销毁;
  6. uv run mcp dev 先行测试,uv run mcp install 接入客户端;
  7. 远程部署按 MCP 基线控制自检,本地 stdio 也需排查 RCE 向量。

遵循这套方法论产出的服务器具备四个特征——类型安全、健壮、文档完整、易于被 LLM 高效使用,这正是 Agent 定义文件结尾对该专家角色的最终定位,也是任何希望以 Python 为 LLM 提供工具能力的开发者可以直接复用的最佳实践基线。

登录后查看全文
awesome-copilot