《MCP + Pydantic + FastAPI：构建可审计、可治理的 AI 工具链——测开工程师的工程化落地指南》

沈宥

发布于 2026-01-08 10:15:09

1200

一、引言：为什么测开要关心 MCP？

在上一篇《FastAPI + MCP 实战解析》中，我们揭示了“AI 直接生成 SQL”这类野蛮接入方式带来的安全、权限、审计等系统性风险，并提出了 MCP（Model Context Protocol） 作为工程级治理方案的核心思想：让 AI 看得见能力，但摸不着系统。

然而，很多读者反馈：“道理我懂，但具体怎么用？Pydantic 怎么配合？如何做工具调用日志？权限校验放哪一层？”——这正是本文要解决的问题。

测开工程师的独特优势在于：既懂业务逻辑，又熟悉自动化与可观测性。 我们不是单纯调模型的人，而是 AI 能力的“守门人”和“调度员”。因此，本文将从工具设计 → 权限控制 → 审计追踪 → 异常兜底 → 多环境隔离五个维度，手把手教你构建一个生产可用的 MCP 工具链。

二、核心组件再认识：Pydantic 是 MCP 的“契约基石”

MCP 的核心是 Tool Schema ——即模型能理解的“能力说明书”。而 FastAPI 生态中，Pydantic 模型天然就是 Schema 的最佳载体。

2.1 为什么必须用 Pydantic？

✅ 强类型校验：防止 order_id="abc" 这种非法输入穿透到数据库。
✅ 自动生成 OpenAPI / JSON Schema：MCP Client 可直接解析。
✅ 序列化/反序列化安全：避免 eval、exec 等危险操作。
✅ 支持嵌套结构：复杂参数也能清晰表达。

📌 关键认知：MCP Tool 的输入输出，必须是 Pydantic BaseModel 的子类，不能是 dict 或 str！

2.2 示例：一个“安全”的订单查询工具

1from pydantic import BaseModel, Field
2from fastapi_mcp import FastApiMCP
3from fastapi import Depends
4
5class OrderQueryInput(BaseModel):
6    order_id: str = Field(..., min_length=6, max_length=32, description="订单ID，由数字和字母组成")
7
8class OrderQueryOutput(BaseModel):
9    order_id: str
10    status: str
11    pay_status: str
12    user_id: str  # 注意：这里故意不返回敏感字段如手机号、地址
13
14@mcp.tool()
15def query_order_status(input: OrderQueryInput) -> OrderQueryOutput:
16    # 此处 input 已经是经过 Pydantic 校验的合法对象
17    order = db.get_order(input.order_id)
18    if not order:
19        raise ValueError("订单不存在")  # MCP 会捕获并返回给模型
20    return OrderQueryOutput(
21        order_id=order.id,
22        status=order.status,
23        pay_status=order.pay_status,
24        user_id=order.user_id  # 仅返回必要字段
25    )

💡 测开视角：这个 OrderQueryOutput 就是我们对 AI 的“数据出口白名单”。任何不在其中的字段，AI 永远拿不到。

三、权限控制：别让 AI 成为“越权访问者”

MCP 最大的价值之一，是 把权限判断从 Prompt 移回后端代码。但具体怎么做？

3.1 方案一：基于用户上下文的权限校验（推荐）

假设你的系统通过 JWT 传递用户身份，可在 Tool 中注入当前用户：

1from fastapi import Request
2
3def get_current_user(request: Request) -> str:
4    # 从 request.headers 解析 token，获取 user_id
5    return "user_123"
6
7@mcp.tool()
8def query_order_status(
9    input: OrderQueryInput,
10    current_user: str = Depends(get_current_user)
11) -> OrderQueryOutput:
12    order = db.get_order(input.order_id)
13    if order.user_id != current_user:
14        raise PermissionError("无权查看他人订单")
15    return OrderQueryOutput(...)

✅ 优势：权限逻辑与业务逻辑解耦，且可复用现有鉴权体系。 ⚠️ 注意：MCP Client 必须在调用时携带有效 Token（可通过中间件自动注入）。

3.2 方案二：基于角色的能力暴露（多模型场景）

如果你有多个 AI Agent（如客服 Agent、运维 Agent），可为不同角色注册不同 Tool：

1# 客服角色只能查订单状态
2customer_service_mcp = FastApiMCP(app, prefix="/mcp/customer")
3@customer_service_mcp.tool()
4def query_order_status(...): ...
5
6# 运维角色可查服务器状态
7ops_mcp = FastApiMCP(app, prefix="/mcp/ops")
8@ops_mcp.tool()
9def check_server_health(...): ...

🔒 安全边界：不同前缀对应不同 API 路由，可通过 Nginx 或网关做 RBAC 控制。

四、审计与可观测性：回答“谁让 AI 干了什么？”

没有审计的 AI 调用等于“黑盒操作”。MCP 必须记录：

谁（User/Agent）调用了哪个 Tool？
输入参数是什么？
执行结果是否成功？
耗时多少？

4.1 实现审计日志（结合 Structlog）

1import structlog
2
3logger = structlog.get_logger()
4
5@mcp.tool()
6def query_order_status(input: OrderQueryInput, current_user: str = Depends(get_current_user)):
7    start = time.time()
8    try:
9        order = db.get_order(input.order_id)
10        if order.user_id != current_user:
11            raise PermissionError("越权")
12        result = OrderQueryOutput(...)
13        logger.info("mcp_tool_call_success",
14                    tool="query_order_status",
15                    user=current_user,
16                    input=input.dict(),
17                    output=result.dict(),
18                    duration_ms=(time.time() - start) * 1000)
19        return result
20    except Exception as e:
21        logger.error("mcp_tool_call_failed",
22                     tool="query_order_status",
23                     user=current_user,
24                     input=input.dict(),
25                     error=str(e))
26        raise

4.2 测开如何利用这些日志？

监控异常调用：如某用户频繁查询不存在的订单 ID（可能在探测）。
性能分析：慢查询告警，优化 DB 索引。
行为回溯：当用户投诉“AI 泄露了我的订单”，可精准定位调用链。

📊 建议：将日志接入 ELK 或 Grafana Loki，建立 MCP 调用大盘。

五、异常处理与兜底策略：AI 不是神，会犯错

MCP Tool 必须考虑以下异常场景：

异常类型	处理建议
输入校验失败（Pydantic 抛出）	自动返回错误，无需额外处理
业务逻辑异常（如订单不存在）	抛出明确异常，MCP 会转为模型可读消息
第三方服务超时	设置 timeout，返回“服务暂时不可用”
数据库连接失败	触发告警，返回友好提示

5.1 统一异常处理器（FastAPI Middleware）

1from fastapi.responses import JSONResponse
2
3@app.exception_handler(ValueError)
4async def value_error_handler(request, exc):
5    return JSONResponse(
6        status_code=400,
7        content={"error": str(exc)}
8    )
9
10@app.exception_handler(PermissionError)
11async def permission_error_handler(request, exc):
12    return JSONResponse(
13        status_code=403,
14        content={"error": "权限不足"}
15    )

✅ 效果：无论 Tool 内抛出什么异常，MCP Client 都能收到结构化错误，而非 500。

六、多环境隔离：开发、测试、生产各司其职

MCP 工具链必须支持环境差异化：

环境	能力暴露	数据源	限流策略
开发	全量 Tool	Mock DB	无限制
测试	全量 Tool	Test DB	轻度限流
生产	白名单 Tool	Prod DB	严格限流+熔断

6.1 实现方式：配置驱动

1# config.py
2class Settings:
3    ENV: str = os.getenv("ENV", "dev")
4    MCP_TOOLS_ENABLED: list = json.loads(os.getenv("MCP_TOOLS_ENABLED", "[]"))
5
6settings = Settings()
7
8# 注册 Tool 时动态判断
9if "query_order" in settings.MCP_TOOLS_ENABLED:
10    @mcp.tool()
11    def query_order_status(...): ...

6.2 测试环境专属 Tool（用于自动化验证）

1# 仅在测试环境暴露
2if settings.ENV == "test":
3    @mcp.tool()
4    def reset_test_data():
5        """重置测试数据，供 E2E 测试使用"""
6        db.truncate_orders()
7        return {"status": "ok"}

🧪 测开价值：你的自动化测试脚本可以直接调用 reset_test_data，无需绕道 UI 或手动清理。

七、扩展场景：不止于查询，还能做什么？

MCP 不只是“只读查询”，合理设计下可支持：

7.1 安全的写操作（需人工审批）

1@mcp.tool()
2def request_refund(input: RefundRequest):
3    # 不直接退款！而是创建审批工单
4    ticket_id = create_approval_ticket(
5        type="refund",
6        data=input.dict(),
7        requester=current_user
8    )
9    return {"ticket_id": ticket_id, "status": "pending_approval"}

✅ 符合治理原则：AI 发起请求，人类决策执行。

7.2 多 Tool 协同排障

1# Step 1: 查订单
2order = query_order_status(order_id)
3
4# Step 2: 查支付流水
5payment = query_payment_log(order.payment_id)
6
7# Step 3: 生成排查报告
8report = generate_troubleshooting_report(order, payment)