可离线阅读 · Python 工程实践

MCP 是让 AI 应用以统一方式发现并调用外部能力的开放协议。本文将协议对象、Python 框架选择、服务端/客户端代码、传输方式、安全边界和部署流程串成一条可落地路径。

MCP Server · FastMCP / 官方 SDK · stdio / Streamable HTTP · 可复制代码

目录

• 一页理解
• 核心对象
• 传输与架构
• Python 框架
• 最小可运行服务
• Python Client
• HTTP 服务
• 完整项目模板
• 安全与生产
• 调试排错
• 上线清单

一页理解

步骤	要做什么	为什么
1	建立能力边界	服务端把能力暴露为工具、资源与提示词；模型并不直接拥有数据库、文件或网络权限。
2	建立会话	客户端通过标准传输连接 Server，初始化后列出能力，再按用户授权调用。
3	接入宿主	桌面客户端、IDE、Agent、Web 后端都可以是 Host；MCP 不绑定某个模型厂商。

核心对象：MCP 解决的是什么问题

可以将 MCP 理解成 AI 应用的“外设协议”。它统一能力发现、调用、上下文读取和交互流程，而不是替代模型、工作流引擎或业务 API。

Host  →  MCP Client  →  MCP Server  →  文件 / 数据库 / 内部 API / SaaS

角色	职责
Host	用户实际使用的应用。负责授权、选择模型、管理一个或多个 Client。
MCP Client	连接并初始化会话；读取 tools/resources/prompts；将调用请求发往 Server。
MCP Server	以声明式元数据暴露业务能力；在自己的权限域内访问外部系统。

对象	适合承载的内容	典型调用	设计建议
Tool 工具	有明确输入、可能产生计算或副作用的动作	查询订单、创建工单、执行只读 SQL	名称用动词；Schema 严格；副作用必须显式确认。
Resource 资源	可定位、可读取的上下文内容	文档、配置、当前项目状态	使用稳定 URI；适合“读”，不是“做”。
Prompt 提示词	可复用的任务模板与参数	代码审查、日报生成、故障分诊	把参数和业务 SOP 写清楚。
Sampling / Elicitation	Server 请求模型生成，或请求用户补充信息	需要模型归纳、缺少审批理由	由 Host 决定是否允许；不可绕过用户控制。

[!WARNING]
MCP 是协议，不是模型自己“学会了访问系统”。每个实际操作仍由 Server 代码和运行环境权限决定。因此，工具描述、参数校验、认证和审计不可省略。

传输方式与适用场景

协议层通常使用 JSON-RPC 风格消息；传输层决定进程如何连接。新项目优先 stdio 或 Streamable HTTP，SSE 仅在兼容已有客户端时保留。

方式	连接形态	最适合	优点	注意事项
stdio	Host 启动本地子进程，通过标准输入输出通信	本地开发、桌面应用、IDE 集成	无需端口、隔离清晰、部署简单	日志必须写 stderr；stdout 只能输出协议消息。
Streamable HTTP	远端 HTTP 端点，支持请求/响应与服务端消息流	共享服务、容器化、企业内网	适配认证、负载均衡、监控、横向扩展	处理 Token、Origin 校验、会话与代理超时。
SSE（旧）	HTTP + Server-Sent Events 的早期实现	必须兼容旧客户端的遗留环境	已有生态可能仍然可用	新服务不应优先选择；确认版本兼容。

两种常见链路

本地 stdio：Host 读取本地配置 → 启动 python server.py 子进程 → 初始化 → 发现工具 → 模型选择工具 → Client 调用 → Server 返回结构化结果。

远端 HTTP：Host 带用户身份令牌访问 HTTPS 网关 → 网关转发 MCP 端点 → Server 将身份映射为业务权限 → 调用内部系统 → 记录审计日志并返回结果。

Python 框架怎么选

协议开发主线是官方 mcp SDK。FastMCP 提供装饰器式的高层体验。复杂 HTTP 服务可以集成 FastAPI/Starlette，但不要为了简单 stdio Server 引入额外 Web 框架。

方案	定位	适合场景	推荐	选择理由
官方 Python SDK（mcp）	协议参考实现与基础能力	细控会话、Client、传输或扩展	首选基础	贴近协议；客户端和服务端能力完整。
FastMCP	类型标注和装饰器式高层框架	绝大多数工具/资源/提示词服务	首选效率	自动从函数签名生成输入 Schema，代码量少。
FastAPI / Starlette	Web 服务框架	已有 API 平台、复杂认证、中间件	按需	复用企业 Web 基础设施，采用其 ASGI 集成方式。
Pydantic	输入/输出模型和校验	嵌套参数、强约束、DTO 复用	建议	把模型生成参数当不可信输入，先校验再执行。
LangChain / LlamaIndex	Agent、检索、工作流生态	Host 侧编排，或把 MCP 工具接入 Agent	可选	不是 MCP Server 的必要依赖；先明确谁是 Host。

[!TIP]
几个本地工具时用 FastMCP + stdio；需要 Python Client 时用官方 SDK；多团队远程复用时用 Streamable HTTP，再补认证、审计和限流。安装时固定兼容版本，并以当前官方文档为准。

最小可运行服务：FastMCP + stdio

这个服务提供只读计算工具、资源和提示词；不需要开放端口，适合先验证 MCP 心智模型与本地集成。

1. 创建独立环境

每个 MCP Server 建议独立虚拟环境，避免 Host 找错解释器或依赖。

mkdir mcp-weather-demo && cd mcp-weather-demo
python -m venv .venv
source .venv/bin/activate       # Windows: .venv\Scripts\activate
python -m pip install --upgrade pip
pip install fastmcp pydantic

2. 编写 server.py

函数签名、类型标注和文档字符串会成为模型理解工具的重要输入。示例刻意只做确定性、无副作用的操作。

from fastmcp import FastMCP
from pydantic import BaseModel, Field

mcp = FastMCP("天气演示服务")


class WeatherQuery(BaseModel):
    city: str = Field(min_length=1, max_length=40, description="要查询的城市")
    unit: str = Field(default="c", pattern="^(c|f)$", description="温度单位：c 或 f")


@mcp.tool()
def get_weather(query: WeatherQuery) -> dict:
    """查询演示天气数据。只用于示范，不连接真实气象服务。"""
    samples = {
        "北京": {"temp_c": 26, "condition": "晴"},
        "上海": {"temp_c": 29, "condition": "多云"},
    }
    data = samples.get(query.city)
    if data is None:
        return {"ok": False, "message": f"没有 {query.city} 的演示数据"}

    temp = data["temp_c"] if query.unit == "c" else round(data["temp_c"] * 9 / 5 + 32, 1)
    return {
        "ok": True,
        "city": query.city,
        "temperature": temp,
        "unit": query.unit,
        "condition": data["condition"],
        "source": "demo",
    }


@mcp.resource("weather://cities")
def available_cities() -> str:
    """可查询城市列表。"""
    return "北京, 上海"


@mcp.prompt()
def weather_brief(city: str) -> str:
    """生成天气简报任务模板。"""
    return f"请查询 {city} 的天气，并用不超过三句话给出出行建议。"


if __name__ == "__main__":
    mcp.run()  # 默认 stdio；不要向 stdout 打印普通日志

[!NOTE]
Server 应由 MCP Host 或 Inspector 发起初始化和调用。stdio 场景中直接运行终端通常看不到普通输出是正常的；不要往 stdout 加 print，否则会破坏协议数据流。

Python Client：以子进程连接本地 Server

Client 的职责是建立会话、初始化、发现工具和调用工具。下例使用官方 SDK；不同小版本导入路径可能略有调整，请以已安装版本文档为准。

# client.py
import asyncio
import sys

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


async def main() -> None:
    params = StdioServerParameters(
        command=sys.executable,
        args=["server.py"],
        env=None,  # 需要传递最小环境变量时显式写在这里
    )

    async with stdio_client(params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            tools = await session.list_tools()
            print("可用工具:", [tool.name for tool in tools.tools])

            result = await session.call_tool(
                "get_weather",
                arguments={"query": {"city": "北京", "unit": "c"}},
            )
            for item in result.content:
                print(item)


if __name__ == "__main__":
    asyncio.run(main())

给 Agent 接入时

1. 调用 list_tools 得到 Schema，再转换为 Agent 框架所需的工具定义。
2. 执行前做用户授权，执行后保留工具名、参数摘要、身份、耗时和结果状态。
3. 工具可能失败、超时、权限不足或返回不完整数据。Client 应区分可重试网络错误与不可重试业务错误，并返回模型可理解的错误。

从本地到远端：Streamable HTTP 使用样式

远程服务应以 HTTPS、身份认证、权限映射和可观测性为默认前提。FastMCP 常可直接以 HTTP transport 运行；具体 CLI 参数以当前安装版本的帮助和文档为准。

# 同一个 server.py；示意命令，先执行 fastmcp --help 核对当前版本参数
fastmcp run server.py --transport streamable-http --host 0.0.0.0 --port 8000

# 生产部署：只将 HTTPS 网关暴露给外部
# https://mcp.example.com/mcp  ->  127.0.0.1:8000/mcp

层	必须处理的事	示例
边缘网关	TLS、Origin/Host 校验、请求体大小限制、限流、WAF	Nginx / API Gateway / Ingress
认证层	验证 OAuth access token 或企业 SSO；长期密钥不可交给模型	JWT 验签、OIDC、mTLS
授权层	将调用者身份映射为数据范围与可用工具	用户只能读取所属项目
业务层	参数校验、幂等性、超时、失败隔离、审计	工单创建使用 idempotency key

[!CAUTION]
不要将“任意 Shell 命令”“任意 SQL”“任意 URL 请求”“任意文件读写”包装成一个工具直接交给模型。应改成小而具体的业务工具，并在服务端执行白名单、参数约束和权限检查。

完整项目模板：可交付的业务 MCP Server

示例以“项目工单查询与创建”为边界：查询可自动执行；创建有副作用，要求幂等键和明确确认标记。数据库/API 细节由你的业务实现替换。

project-mcp/
├── pyproject.toml
├── .env.example                 # 只放变量名，不放真实密钥
├── src/project_mcp/
│   ├── server.py                # MCP 工具、资源、提示词注册
│   ├── service.py               # 业务 API / DB 调用
│   ├── auth.py                  # 身份与权限上下文
│   └── models.py                # Pydantic 入参/出参
└── tests/
    ├── test_tools.py
    └── test_service.py

# src/project_mcp/server.py
from fastmcp import FastMCP
from pydantic import BaseModel, Field

from project_mcp.service import create_ticket, list_tickets

mcp = FastMCP("项目工单服务")


class ListTicketsInput(BaseModel):
    project_id: str = Field(min_length=1, max_length=64)
    status: str | None = Field(default=None, description="可选：open、closed")


class CreateTicketInput(BaseModel):
    project_id: str = Field(min_length=1, max_length=64)
    title: str = Field(min_length=3, max_length=120)
    description: str = Field(min_length=1, max_length=4000)
    idempotency_key: str = Field(min_length=16, max_length=128)
    confirmed: bool = Field(description="用户已确认创建，必须为 true")


@mcp.tool()
async def list_project_tickets(input: ListTicketsInput) -> dict:
    """查询项目工单。只读操作，可用于总结待办和排查问题。"""
    tickets = await list_tickets(project_id=input.project_id, status=input.status)
    return {"ok": True, "items": tickets, "count": len(tickets)}


@mcp.tool()
async def create_project_ticket(input: CreateTicketInput) -> dict:
    """创建项目工单。有副作用，仅在用户明确确认后调用。"""
    if not input.confirmed:
        return {"ok": False, "error": "需要用户明确确认后才能创建工单"}

    ticket = await create_ticket(**input.model_dump(exclude={"confirmed"}))
    return {"ok": True, "ticket": ticket}


if __name__ == "__main__":
    mcp.run()

# tests/test_tools.py：验证副作用边界
import pytest

from project_mcp.server import CreateTicketInput, create_project_ticket


@pytest.mark.asyncio
async def test_create_rejects_unconfirmed_request():
    result = await create_project_ticket(CreateTicketInput(
        project_id="demo",
        title="修复登录失败",
        description="复现步骤已附上",
        idempotency_key="a" * 16,
        confirmed=False,
    ))
    assert result == {"ok": False, "error": "需要用户明确确认后才能创建工单"}

安全、权限与生产治理

模型生成的每个参数都应视为外部不可信输入。MCP 带来的不是绕过权限，而是将身份、授权和审计延伸到模型可调用的边界。

工具设计规则

• 一个工具只表达一个清楚的业务动作，不提供万能执行器。
• 以 Pydantic/JSON Schema 限制长度、枚举、格式和嵌套对象。
• 读取与写入分开；删除、支付、发布要求二次确认。
• 返回最小必要数据，不泄露凭证、全量个人信息或内部错误栈。

运行时控制

• 每次调用绑定最终用户或服务身份，不共享超权限机器人账号。
• 设置连接、下游请求、工具执行总超时与并发上限。
• 记录 trace_id、调用者、工具、参数摘要、授权结果、耗时和状态。
• 密钥由环境或密钥系统注入；日志对 token、密码、PII 脱敏。

风险	说明	控制措施
提示注入	资源文本、网页、工单内容也可能含恶意指令	将它们当数据，不允许其改写工具权限或系统策略。
最小权限	一个万能 token 的爆炸半径很大	每个 Server 使用独立服务账号；按环境、项目和操作拆分角色。
供应链	第三方 MCP Server 可能读取或发送敏感数据	审核源码、依赖、网络出口和数据处理方式；固定依赖版本并扫描漏洞。

调试、测试与常见故障

调试顺序从“协议有没有建立”到“工具能否执行”再到“业务权限是否正确”，不要一开始归因于模型。

1. 先验证 Server 能被启动：确认 Host 配置的 Python 绝对路径、工作目录、虚拟环境和环境变量。stdio 日志写 stderr。
2. 检查初始化与能力发现：用 MCP Inspector 或自写 Client 调用 initialize、list_tools、list_resources，确认工具名和 Schema 出现。
3. 绕开模型直接调用工具：以固定参数验证输入校验、下游 API、权限、超时和返回结构，先得到可重复的确定性结果。
4. 最后检查 Agent 行为：模型是否理解工具说明、是否遗漏确认步骤、是否正确消费结果。这一层才属于提示词或编排问题。

现象	常见原因	处理
Host 启动失败	解释器路径错误、未激活依赖、相对路径依赖工作目录	用绝对路径；在 Host 同一环境手动执行；固定工作目录。
工具列表为空	装饰器未加载、导入报错、没有完成 initialize	检查 stderr；先用最小 Client 列工具；确保注册代码启动即执行。
stdio 解析错误	stdout 输出了 print、logging 或库 banner	stdout 只留协议；日志配置到 stderr。
HTTP 401/403	Token 未透传、Scope 不匹配、网关剥离 Header	从网关到服务逐跳验证身份；不要用 query string 传 token。
模型重复或误调用	工具描述模糊、缺少确认状态、返回不完整	收紧 Schema 和文档字符串；将副作用前置条件做成必填字段。

从零到可用的交付清单

开发完成前

• 每个工具有清楚名称、文档字符串、类型和 Schema。
• 直接调用测试覆盖正常、非法、权限拒绝和下游失败。
• 副作用工具具备确认、幂等性和审计信息。
• 密钥不进入代码、示例、日志或工具返回值。
• stdio 日志不污染 stdout。

发布完成前

• HTTP 场景启用 TLS、认证、授权、限流和请求大小限制。
• 服务账号遵循最小权限，环境和租户隔离明确。
• 具备健康检查、超时、错误监控、调用链与脱敏日志。
• 固定 SDK/依赖版本，升级前验证 Host 兼容性。
• 交付连接方式、权限范围、工具清单和示例参数。

[!TIP]
推荐落地路线：先用 FastMCP 做只读 stdio Server → 用官方 Python Client 写一条可重复集成测试 → 再把真实业务权限、审计和 HTTP 部署接上。每次只增加一个能力边界，问题会更容易定位。

---

本文覆盖通用 MCP 工程模式。MCP 规范与 Python SDK 迭代较快，执行安装与部署时请固定版本，并以对应版本的官方文档、CLI --help 和集成测试结果为准。