AI Agent原理篇13：MCP 实战：写一个自己的 MCP Server

上一篇讲了 MCP 的原理，这一篇动手写一个。目标是把一个业务数据库的查询能力做成 MCP Server，让任何支持 MCP 的 Agent 应用都能直接使用。

说明：示例基于官方 Python SDK（mcp 包）2.x 版本。1.x 版本中同样的类叫 FastMCP，导入方式是 from mcp.server.fastmcp import FastMCP，用法基本相同。SDK 更新较快，接口如有变化，以官方文档为准，思路是一样的。

一、准备

pip install "mcp[cli]"

为了让示例能直接运行，我们用 Python 自带的 SQLite 建一个演示用的订单表。实际项目中把连接换成 MySQL 或其他数据库即可。

# init_db.py：生成演示数据
import sqlite3, random, datetime

conn = sqlite3.connect("shop.db")
conn.execute("""CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY, created_at TEXT, channel TEXT,
    amount REAL, status TEXT)""")
now = datetime.datetime.now()
for i in range(2000):
    t = now - datetime.timedelta(minutes=random.randint(0, 7 * 24 * 60))
    conn.execute("INSERT INTO orders (created_at, channel, amount, status) VALUES (?,?,?,?)",
                 (t.strftime("%Y-%m-%d %H:%M"), random.choice(["web", "app"]),
                  round(random.uniform(10, 500), 2),
                  random.choice(["paid", "paid", "paid", "failed"])))
conn.commit()

二、最小的 MCP Server

# shop_server.py
import sqlite3
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("shop-data")
DB = "shop.db"


def query(sql, params=()):
    # 以只读模式打开数据库，从根本上杜绝写操作
    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    try:
        return [dict(r) for r in conn.execute(sql, params).fetchall()]
    finally:
        conn.close()


@mcp.tool()
def orders_by_hour(start: str, end: str, channel: str = "all") -> list[dict]:
    """按小时统计订单数量和金额。

    用于分析下单量的变化趋势。时间格式 YYYY-MM-DD HH:MM。
    channel 可选 web、app、all。时间范围建议不超过 7 天。
    """
    sql = """SELECT substr(created_at, 1, 13) AS hour,
                    COUNT(*) AS orders, ROUND(SUM(amount), 2) AS amount
             FROM orders WHERE created_at BETWEEN ? AND ?"""
    params = [start, end]
    if channel != "all":
        sql += " AND channel = ?"
        params.append(channel)
    sql += " GROUP BY hour ORDER BY hour"
    return query(sql, params)


@mcp.tool()
def failed_order_rate(start: str, end: str) -> dict:
    """统计时间范围内的订单失败率，用于判断支付环节是否异常。"""
    rows = query("""SELECT COUNT(*) AS total,
                           SUM(status = 'failed') AS failed
                    FROM orders WHERE created_at BETWEEN ? AND ?""", (start, end))
    total, failed = rows[0]["total"], rows[0]["failed"] or 0
    return {"total": total, "failed": failed,
            "failed_rate": round(failed / total, 4) if total else None}


@mcp.resource("schema://orders")
def orders_schema() -> str:
    """orders 表的结构说明"""
    return ("orders(id, created_at 下单时间, channel 渠道 web/app, "
            "amount 金额, status 状态 paid/failed)")


if __name__ == "__main__":
    mcp.run()   # 默认使用 stdio 传输

几个要点：

函数的类型注解自动变成参数 Schema，str、int、默认值都会被识别。
函数的文档字符串就是工具的 description。第 05 篇讲的 description 写法在这里同样适用：做什么、何时用、有什么限制。
返回值会自动序列化，返回字典或列表即可。
只读连接：数据库用只读模式打开。实际项目中更推荐给 MCP Server 单独建一个只读数据库账号。

三、本地调试

SDK 自带一个调试工具（MCP Inspector），可以在浏览器里查看工具列表、手动调用工具：

mcp dev shop_server.py

先在这里确认每个工具的参数和返回值都正确，再接入 Agent 应用。

四、接入 Agent 应用（stdio 方式）

支持 MCP 的桌面应用和编码工具，通常都有一个配置文件，写明怎么启动 Server。格式大致如下：

{
  "mcpServers": {
    "shop-data": {
      "command": "python",
      "args": ["/path/to/shop_server.py"]
    }
  }
}

应用启动时会运行这个命令，通过标准输入输出与 Server 通信。之后你问「最近一天哪个小时下单量最低」，Agent 就会调用 orders_by_hour 去查。

stdio 方式的特点：Server 跑在用户本机，用的是本机的权限和网络，不需要鉴权，但每个用户都要自己配置一份。

五、远程部署（Streamable HTTP 方式）

如果希望团队共用一个 Server，就把它部署成 HTTP 服务：

if __name__ == "__main__":
    mcp.run(transport="streamable-http")   # 默认监听 8000 端口，路径 /mcp

生产环境通常放在 Nginx 后面：

location /mcp {
    proxy_pass http://127.0.0.1:8000/mcp;
    proxy_http_version 1.1;
    proxy_buffering off;          # 流式响应需要关闭缓冲
    proxy_read_timeout 300s;
}

注意：Server 默认只监听本机，并会检查请求的 Host 头，防止 DNS 重绑定攻击。所以 Nginx 里不要把 Host 改成外部域名，保持默认转发即可；如果确实需要，按 SDK 文档配置允许的主机名。

客户端配置改为填写 URL，比如 https://your-domain/mcp。

六、鉴权与安全

远程 Server 暴露在网络上，必须做好保护：

鉴权：正式方案是按 MCP 规范实现 OAuth 授权，用户以自己的身份登录授权。内部小范围使用时，也可以先用 Nginx 层的访问控制或固定令牌，但这只是过渡办法。
最小权限：只读账号、只暴露必要的表、限制单次查询的数据量和时间范围。
不要提供「执行任意 SQL」的工具：看起来灵活，实际上等于把数据库交给了模型，也给注入攻击留了口子。用具体的业务工具代替。
记录调用日志：谁、在什么时间、调用了什么工具、参数是什么，出了问题能追查。

七、设计工具时的取舍

问题	建议
一个通用查询工具，还是多个专用工具？	多个专用工具，每个对应一个业务问题，更安全，模型也更容易选对
返回原始明细，还是汇总数据？	默认返回汇总；需要明细时提供单独的工具，并限制条数
表结构放 tool 还是 resource？	相对静态的说明信息适合做成 resource，供应用按需读取
出错时怎么返回？	返回清晰的错误说明，告诉模型怎么改参数，而不是抛出堆栈信息

本篇小结

用 SDK 写 MCP Server 很简单：函数加装饰器，类型注解变成 Schema，文档字符串变成描述。
先用调试工具验证，再接入应用。
stdio 适合个人本地使用，Streamable HTTP 适合团队共享。
远程部署要做鉴权、最小权限、日志记录，不要暴露任意 SQL。

思考与练习

运行本篇的示例，再加一个工具：查询某个时段金额最高的 10 笔订单。注意限制返回条数。
你的 MCP Server 应该暴露哪些只读工具？哪些操作即使有需求也不应该做成工具？

上一篇：AI Agent原理篇12：MCP 协议原理
下一篇：AI Agent原理篇14：Skills：按需加载的能力包
 返回系列总目录