ARTICLE · 1001116
我照着文档从零撸了个MCP Server,真正的坑在版本锁
我照着文档从零撸了个MCP Server,真正的坑在版本锁周一早上,产品扔过来一句话,差点把我咖啡呛出来。 “小张,能不能让公司的 AI 助手自己查一下学员成绩?就是像问 ‘张伟这学期平均分多少’ 这种,别每次都要人先去数据库查了再喂给它。” 我盯着需求看了三秒,脑子里立刻浮现出之前给 Agent 写工具函数的日子:一开始就两三个函数还好说,等加到十几个,每个都要在代码里声明 schema,调用逻辑还跟框架强耦合。同一套查询逻辑,从 LangChain 切到别的框架就得重写一遍。那滋味,谁经历过谁知道。 所以这次听说有个叫 MCP 的玩意儿——Model Context Protocol,可以让模型通过一种标准协议直接去访问数据源,我的第一反应是:终于有人把 USB 接口这事儿做了。道理其实差不多:USB 让电脑和外设不用管对方具体是什么牌子的芯片,插上就能用;MCP 之于 Agent,也是这么个意思。 模型不再是“啥也够不着”的对话机器了,而是通过 MCP Server 这个中间层,能拿到数据库、文件系统、API 里的真实数据。 我手边正好有个沙箱环境,就打算写一个 Python 的 MCP Server,让 Agent 能回答这类问题: 说白了,就是给 LLM 接上我的 SQLite 数据库。将来决定要接真实系统了,底层换 MySQL 即可,上面的工具定义完全不用动。 我一开始以为这玩意儿怎么也得写个几百行协议处理代码。结果了解了一下当前 Python SDK 提供的FastMCP封装——写起来跟 Flask 差不多,核心逻辑浓缩到几十行就完事了。 先说下依赖环境。我机器默认 Python 是 3.9.x,MCP SDK 要求 3.10 及以上,所以直接用uv建了独立环境。 $ uv venv --python 3.12 .venv Using CPython 3.12.13 Creating virtual environment at: .venv 然后在pyproject.toml里把依赖加进去: [project] name = "grade-mcp-server" version = "0.1.0" requires-python = ">=3.10" dependencies = [ "mcp>=1.28,<2", "uvicorn>=0.30.0", ] 注意这个mcp<2,这是我这回踩的第一个坑,一会儿详细讲。 数据层还是很朴素的,就是连着做了个 SQLite 库。为了让 Agent 有数据可查,我麻利地把表结构建出来,顺便塞了条张伟 S001 的种子数据: MCP 的定义方式我在代码里写出来你能感受到,工具层工具定义基本不碰 SQL 逻辑。只需要给每个函数一个清晰的 docstring,因为docstring 就是给大模型看的说明书。这句话我在项目里原样写在注释里,后来发现真不是开玩笑。 你有没有注意到,我每个函数的 docstring 里都把参数格式给出来,像S001、2026-秋。为了让模型少猜。实测只差这一步,准确性能有肉眼可见的差距。 不过你也许奇怪:这个@mcp.tool()装饰器,底层到底做了什么?其实它把这些普通函数转化成了 JSON Schema 描述的工具,这些描述在 MCP 握手的tools/list阶段就发送给客户端。之后 LLM 看到用户问“张伟多少分”,就会根据工具说明,自动组装参数,再调起我的tools/call接口。 MCP 的传输支持两种:本地就是stdio,远程就是StreamableHTTP。FastMCP 的好处是它能用同一个 server 实例切换。 麻烦的是这库先跑 HTTP 再跑 stdio 顺序不同,有些兼容性问题。我这次是先在项目里验证的 stdio。 本地冒烟测试,就是模拟一个客户端把 server 拉起来,跟它对话: 运行结果也正常: 发现工具: - get_student_info: 按学号精确查询学生信息,学号格式如'S001'。 - fuzzy_query_students: ... - get_student_scores: ... [{"name": "数据库原理", "term": "2026-秋", "credits": 4, "score": 88.5}] 说起来不复杂,第一次装依赖我图省事直接uv pip install mcp uvicorn,结果装上的是 alpha 版 SDK,有些导入路径改了。更坑的是我跑去 HTTP 模式做集成测试的时候发现两轮握手之后程序卡死——进程直接挂。 当时吓得我还以为是环境变量问题,折腾半天后来干脆到 PyPI 上翻了下版本,发现最新稳定版是1.x,直接追加约束: uv pip install "mcp>=1.28,<2" 这下整个流程才通顺。以后项目初始化就老实锁版本,不搞那套“能跑就行”了。 最后还是要部署到内网。我先起了 FastMCP 的 HTTP 模式: 系统服务用 systemd 托管,老一套了: 注意WorkingDirectory别乱放,否则可能源目录被日志撑爆。 因为 Hermes 原生支持 MCP 配置,我在hermes.toml填了下面这行,重启就自动连上了: [[mcp_servers]] name = "grades" transport = "http" url = "http://127.0.0.1:8100/mcp" 然后发现工具名会被自动加前缀:mcp__grades__get_student_scores。这个前缀再运行时要看清楚,因为工具名称变了,有时候从日志里调试你会觉得很蒙,还以为函数名写错了,其实是服务器名称生成了命名空间。 体验对比下来,如果 Agent 客户端本身就支持 MCP,你不用再给自己的 model 加一个 tools 列表了。而我所做的只是把这些@mcp.tool()函数写好,剩下的全部协议握手、JSON-RPC 数据传输,SDK 都给处理完了。 这次开发给我较深印象的三件事: 第一,docstring 就是给 AI 看的接口文档。你自己可能觉得“这不就是注释吗”,但对大模型来说,它就是理解函数行为的主要手段。以后写 MCP 工具函数,我会把参数格式、边界情况(查不到返回什么、能不能模糊)全写进 docstring。 第二,测试要早点写。不只是数据库 SQL 对错的测试,更要测“模型调用会不会出错”。传错参数、空数据、异常提示都要让模型能读明白,不然调用链很容易断。 第三,MCP 虽香,还是要看场景。对于有外部数据依赖的 Agent,MCP 协议确实是降低后期改动成本的方法之一。但用它来包装一个两个人三个月就丢弃的工具,也确实稍微重了点。 项目代码量不多,大概不到三百行,核心入口逻辑也不复杂。如果手里刚好有接外部数据的活儿,特别是想做一个能够反复给别人 Agent 用的能力,把 MCP Server 当作连接大脑和世界的小接口,真香。 装好试一下。万一你也卡在 SDK 版本问题出不来,欢迎评论区跟我聊聊你踩到啥坑了。
先定个小目标:做个成绩查询的 MCP Server
“查一下 S001 这个学生的所有成绩” “2026-秋学期《数据库原理》的最高分是多少” “张伟的加权平均分算一下”
用 FastMCP 写 MCP Server 是你想象不到的简单
grade_mcp/db.py import os import sqlite3 DB_PATH = os.environ.get("GRADES_DB_PATH", "grades.db") SCHEMA = """ CREATE TABLE IF NOT EXISTS students ( id TEXT PRIMARY KEY, name TEXT NOT NULL, class_name TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS courses ( id TEXT PRIMARY KEY, name TEXT NOT NULL, credits INTEGER NOT NULL DEFAULT 3 ); CREATE TABLE IF NOT EXISTS grades ( id INTEGER PRIMARY KEY AUTOINCREMENT, student_id TEXT NOT NULL REFERENCES students(id), course_id TEXT NOT NULL REFERENCES courses(id), term TEXT NOT NULL, score REAL NOT NULL ); """ def get_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_connection() conn.executescript(SCHEMA) conn.execute( "INSERT INTO students (id, name, class_name) VALUES (?, ?, ?)", ("S001", "张伟", "计科2201"), ) # ... courses 和 grades 种子里我补了几条 conn.commit() conn.close()
核心来了:怎么把函数暴露成 Agent 能用的“工具”
grade_mcp/server.py from mcp.server.fastmcp import FastMCP import grade_mcp.db as db mcp = FastMCP("grades") @mcp.tool() def get_student_info(student_id: str) -> str: """按学号精确查询学生信息,学号格式如 'S001'。 返回内容包含学生姓名和班级。 """ conn = db.get_connection() row = conn.execute("SELECT * FROM students WHERE id = ?", (student_id,)).fetchone() conn.close() return dict(row) if row else "未找到该学生" @mcp.tool() def fuzzy_query_students(name: str) -> str: """根据姓名模糊查询学生列表,如输入'张'匹配'张伟'、'张明明'。 返回格式是 JSON 数组字符串。 """ conn = db.get_connection() rows = conn.execute( "SELECT * FROM students WHERE name LIKE ?", (f"%{name}%",) ).fetchall() conn.close() return [dict(r) for r in rows] @mcp.tool() def get_student_scores(student_id: str) -> str: """查询某学生的全部成绩,返回课程名、学期、学分、成绩。 学生ID必须精确,如'S001'。 """ conn = db.get_connection() rows = conn.execute( """ SELECT c.name, g.term, c.credits, g.score FROM grades g JOIN courses c ON g.course_id = c.id WHERE g.student_id = ? """, (student_id,), ).fetchall() conn.close() return [dict(r) for r in rows]
跑起来:先过本地 stdio 冒烟测试
scripts/test_stdio.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params = StdioServerParameters( command="python", args=["-m", "grade_mcp.server"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await session.list_tools() print("发现工具:") for tool in tools.tools: print(f" - {tool.name}: {tool.description}") result = await session.call_tool( "get_student_scores", {"student_id": "S001"} ) print(result.content[0].text) asyncio.run(main())
踩坑一: 版本锁错了,直接 500
部署上线:远程给别人用
入口部分 if name == "main": import os db.init_db() transport = os.environ.get("MCP_TRANSPORT", "stdio") if transport == "http": # FastMCP 封装了 StreamableHTTP mcp.run(transport="http", host="0.0.0.0", port=8100) else: mcp.run(transport="stdio")
/etc/systemd/system/grade-mcp.service [Unit] Description=Grade MCP Server After=network.target [Service] User=www-data WorkingDirectory=/opt/grade-mcp-server Environment=GRADES_DB_PATH=/var/lib/grade-mcp/grades.db Environment=MCP_TRANSPORT=http ExecStart=/opt/grade-mcp-server/.venv/bin/python -m grade_mcp.server Restart=always RestartSec=3 [Install] WantedBy=multi-user.target
接入 Hermes Agent,看 Agent 自己用起来
吃一堑长一智的个人总结
欢迎关注「火柴人de故事」,一起折腾 AI 和技术。