用 Python 开发 MCP 服务很简单，完整案例！

模型能回答订单为什么失败，但它查不到订单。

数据库里明明有错误码、重试次数、最后更新时间，模型却只能让人“检查网络或者查看日志”。这种回答我一般直接忽略，信息都没拿到，分析得再像也没用。

MCP 解决的就是这层连接问题：把查询订单、读取日志、执行脚本这些能力，按统一协议暴露给 AI 客户端。官方 Python SDK 已经把协议处理、参数校验和工具描述封装好了，开发时主要写业务函数。

下面做一个订单排障服务。模型可以通过它查询订单状态，也可以拉取最近失败的订单。

当前官方仓库仍推荐生产环境使用稳定的 v1.x，v2 还处于预发布阶段，所以依赖别直接放开到最新版，我一般会把大版本卡住。

uv init order-mcp
cd order-mcp
uv add "mcp[cli]>=1.27,<2"

准备一份测试数据 orders.json：

[
  {
    "order_no": "SO20260710001",
    "status": "FAILED",
    "error_code": "INVENTORY_TIMEOUT",
    "retry_count": 3,
    "updated_at": "2026-07-10 09:42:18"
  },
  {
    "order_no": "SO20260710002",
    "status": "PAID",
    "error_code": "",
    "retry_count": 0,
    "updated_at": "2026-07-10 09:45:06"
  }
]

服务端代码放在 server.py：

import json
import logging
from pathlib import Path
from typing import Any
from mcp.server.fastmcp import FastMCP
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
DATA_FILE = Path(__file__).with_name("orders.json")
mcp = FastMCP(
    "order-ops",
    json_response=True,
)

def load_orders() -> list[dict[str, Any]]:
    try:
        content = DATA_FILE.read_text(encoding="utf-8")
        data = json.loads(content)
    except FileNotFoundError as exc:
        raise RuntimeError(f"订单数据文件不存在：{DATA_FILE}") from exc
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"订单数据格式损坏：第 {exc.lineno} 行") from exc
    ifnot isinstance(data, list):
        raise RuntimeError("订单数据必须是 JSON 数组")
    return data

@mcp.tool()
def find_order(order_no: str) -> dict[str, Any]:
    """根据订单号查询状态、错误码、重试次数和更新时间。"""
    normalized_no = order_no.strip().upper()
    ifnot normalized_no:
        raise ValueError("订单号不能为空")
    logging.info("query order order_no=%s", normalized_no)
    for order in load_orders():
        if order.get("order_no") == normalized_no:
            return {
                "found": True,
                "order": order,
                "advice": build_advice(order),
            }
    return {
        "found": False,
        "order_no": normalized_no,
        "message": "没有查到该订单，检查订单号或数据同步状态",
    }

@mcp.tool()
def list_failed_orders(limit: int = 5) -> list[dict[str, Any]]:
    """查询最近失败的订单，limit 取值范围为 1 到 20。"""
    ifnot1 <= limit <= 20:
        raise ValueError("limit 必须在 1 到 20 之间")
    failed = [
        order
        for order in load_orders()
        if order.get("status") == "FAILED"
    ]
    failed.sort(
        key=lambda item: item.get("updated_at", ""),
        reverse=True,
    )
    return failed[:limit]

def build_advice(order: dict[str, Any]) -> str:
    error_code = order.get("error_code")
    if error_code == "INVENTORY_TIMEOUT":
        return"先查库存接口耗时和线程池队列，不要直接重放订单"
    if error_code == "PAYMENT_REJECTED":
        return"检查支付渠道返回码，确认后再决定是否允许重试"
    if order.get("status") == "PAID":
        return"订单已支付，无需补偿"
    return"当前错误码没有匹配处理规则，需要结合业务日志继续排查"

@mcp.resource("runbook://order-failure")
def order_failure_runbook() -> str:
    """返回订单失败的排查顺序。"""
    return"""
1. 核对订单状态和最后更新时间
2. 检查 error_code，确认是否允许重试
3. 查询对应下游接口的 trace
4. 检查线程池、连接池和超时配置
5. 确认幂等后再执行补偿
""".strip()

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

这里没有手写 JSON Schema，也没有自己解析 MCP 请求。

order_no: str、limit: int 这些类型标注会变成工具参数定义，函数的 docstring 会变成工具说明。模型拿到工具列表后，知道工具叫什么、接收什么参数、适合什么时候调用。这个地方确实比自己封一层 HTTP 接口省事。

启动服务：

uv run server.py

默认连接地址是：

http://localhost:8000/mcp

测试时别急着接模型，先用 MCP Inspector 把工具单独跑通：

npx -y @modelcontextprotocol/inspector

在 Inspector 中填入服务地址，然后调用：

find_order
order_no = SO20260710001

正常会返回订单信息以及一条排查建议。再调用 list_failed_orders，就能拿到最近的失败订单。

有个坑得单独提一下。

如果改成 stdio 方式运行，业务日志不能随手 print() 到标准输出。标准输出走的是 MCP 的 JSON-RPC 消息，混进去一行调试文字，客户端可能直接报协议解析失败。日志写到 stderr 或使用 Python 的 logging，官方文档也明确提醒了这一点。

代码跑起来只是第一步。真放到线上，查询数据库的账号只能给只读权限，删除、退款、补偿这类工具必须做鉴权、审计和参数限制。

MCP 降低的是接入成本，不是把危险操作变安全了。工具名字可以随便起，权限边界不能随便画。

以上就是“用 Python 开发 MCP 服务很简单，完整案例！”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。