第 19 篇把 MCP(Model Context Protocol)的协议层拆完了:三原语 Tools/Resources/Prompts、两种传输 stdio 与 Streamable HTTP、底下跑的 JSON-RPC 消息流。但协议只是纸面约定——工具怎么变成一个独立进程、独立机器上的服务,Host 又怎么把它拉起来?这一篇动手把答案写出来:从零到一个能被 Claude Desktop、Claude Code 或自建 Agent 直接调用的 MCP Server。
本篇是《AI Agent 开发》系列第 20 篇,代码基于官方 MCP Python SDK 2.3.x(2026-10 的当前稳定版本),环境沿用第 2 篇的 uv 项目。全文按"安装避坑 → 第一个 Server → 工具进阶 → 资源与模板 → SQLite 实战 → 部署形态 → 测试调试"的顺序推进,每段代码都可以直接抄进项目。
1、安装与包名陷阱
官方 SDK 的包名是 mcp,用 uv 装进项目(要求 Python 3.10+,系列基线 3.11):
uv add "mcp[cli]" # 或 pip install "mcp[cli]"
[cli] 这个 extra 一定要带上,它额外装上三个命令:mcp dev(在调试器里跑 Server)、mcp run(直接跑 Server)、mcp install(一键注册进 Claude Desktop),后面都会用到。版本与更新记录见 PyPI 的 mcp 包 与 官方 SDK 仓库。
真正的新手坑不在安装命令,在类名。搜"MCP Python 教程",前排结果大多写的是 from mcp.server.fastmcp import FastMCP——这在 SDK v1 时代是对的:FastMCP 原本是社区项目,2024 年整个合并进了官方 SDK。但 v2(2026-07-28 GA)把它改名为 MCPServer,模块从 mcp.server.fastmcp 挪到了 mcp.server.mcpserver,旧路径在新版下直接抛 ModuleNotFoundError。本篇统一用 v2 写法:
from mcp.server.mcpserver import MCPServer
还有第二个坑:那个社区项目并未停更,如今以 fastmcp 包独立发布(gofastmcp.com),在官方 SDK 之上加了认证、代理、多服务器组合等框架级能力。它是个好项目,但和官方 SDK 是两套 API——pip install fastmcp 装到的是它,本篇的代码在它上面跑不通。
装包认准
uv add mcp;导入认准from mcp.server.mcpserver import MCPServer。看到 FastMCP 字样的教程,先确认它针对的是官方 SDK v1 还是独立的 fastmcp 包,别混着抄。
改名的来龙去脉与全部破坏性变更,官方整理在 v1 到 v2 迁移指南里,从 v1 迁移老服务时值得一读。
2、第一个 Server
新建 server.py,不到 20 行跑通一个时区时间查询服务:
from datetime import datetime
from zoneinfo import ZoneInfo
from mcp.server.mcpserver import MCPServer
mcp = MCPServer("time-query") # 名字会显示在 Host 的服务器列表里
@mcp.tool()
def now(tz: str = "Asia/Shanghai") -> str:
"""查询指定时区的当前时间,如 now(tz="Asia/Tokyo")。"""
try:
return datetime.now(ZoneInfo(tz)).isoformat(timespec="seconds")
except Exception:
return f"未知时区 {tz},请用 IANA 名称,如 Asia/Shanghai"
if __name__ == "__main__":
mcp.run() # 不传参数 = stdio 传输
解读三件事。MCPServer("time-query") 实例化服务,参数是显示名;@mcp.tool() 把普通函数注册为工具——tz: str 的类型注解自动生成参数 schema,docstring 自动成为工具描述,模型全靠这两样决定何时调用、怎么传参;mcp.run() 不传参数就是 stdio 传输。Windows 用户注意:zoneinfo 依赖系统时区库,需补一句 uv add tzdata,Linux/macOS 不用管。
stdio 的运行原理值得单独讲清:Server 是 Host 的子进程。Claude Desktop、Claude Code 或你的 Agent 程序(统称 Host)启动时按配置把 server.py 拉起,子进程的 stdin/stdout 就是协议通道——Host 往 stdin 写一行 JSON-RPC 请求,Server 从 stdout 回一行响应。由此立刻推出一条铁律:
Server 代码里严禁 print()。stdout 上的每个字节都会被当成协议报文,一条 print 就能打挂整条通道;日志一律走 logging 模块,它默认输出到 stderr。
顺带解释一个新手必撞的现象:直接 python server.py 会发现程序挂在原地一动不动。它没有死——它在等 stdin 上到来的 JSON-RPC 请求,而终端不会发。stdio Server 的正确打开方式是被 Host 拉起,或者在下一节用 Inspector 手动当一回 Host。
3、工具定义进阶
第 5 篇讲 LangChain 的 @tool 装饰器时给过两条契约:类型注解生成参数 schema,docstring 成为工具描述。@mcp.tool() 把同一套契约原样搬了过来——从进程内函数到进程外服务,工具定义的写法没变,变的只是执行位置。所以这一节只讲增量:参数模型、返回值、图片。
参数一多就该上 Pydantic 模型,Field 的约束会原样进入 schema:
from pydantic import BaseModel, Field
from mcp.server.mcpserver import MCPServer
mcp = MCPServer("notes")
NOTES = ["MCP 协议详解", "工具调用基础", "RAG 流水线"]
class NoteQuery(BaseModel):
keyword: str = Field(description="搜索关键词")
max_results: int = Field(default=5, ge=1, le=20, description="条数上限")
@mcp.tool()
def search_notes(q: NoteQuery) -> dict:
"""按关键词搜索笔记标题,返回命中列表与总数。"""
hits = [t for t in NOTES if q.keyword in t]
return {"total": len(hits), "items": hits[: q.max_results]}
解读:ge=1, le=20 变成 schema 里的数值边界,越界参数在进函数前就被拦下;返回 dict 时 SDK 生成两份内容——content(给模型看的文本)与 structured_content(给程序用的 JSON),第 21 篇接入 Agent 时直接读后者。str、list、标量、Pydantic 模型的返回值都支持,规则与 dict 相同。另一个 v2 细节:同步 def 的工具函数会被自动丢进工作线程执行,不会卡住事件循环。
要回传图片就用 Image 类型,路径或原始字节都行:
from mcp.server.mcpserver import Image
@mcp.tool()
def trend_chart(days: int) -> Image:
"""生成近 N 天浏览量趋势图。"""
return Image(path=f"charts/trend-{days}.png", format="png")
Image 会被包装成 image content block,多模态 Host(如 Claude Desktop)能直接渲染;不支持图片的 Host 则收到一段降级的文本说明。
4、Resources 与 Prompts
Tools 是"模型主动调用的动作",另外两个原语各管一摊:Resources 是数据面,把文件、配置、元信息这类只读上下文暴露给 Host 按需读取;Prompts 是模板面,参数化的提示词预制件,在 Claude Desktop 里表现为斜杠命令。
from pathlib import Path
from mcp.server.mcpserver import MCPServer
mcp = MCPServer("kb")
@mcp.resource("file:///data/glossary.md")
def glossary() -> str:
"""术语表全文,Host 按需读取后注入上下文。"""
return Path("data/glossary.md").read_text(encoding="utf-8")
@mcp.prompt()
def audit_sql(sql: str, dialect: str = "SQLite") -> str:
"""生成一段 SQL 审查提示词。"""
return f"请从安全与性能两个角度审查这条 {dialect} 语句:\n{sql}"
解读:@mcp.resource 的参数是 URI,Host 靠它寻址,URI 也支持模板参数(如 "memo://{name}",函数签名里的 name 自动接住路径里的值);@mcp.prompt() 的函数返回值就是模板内容,参数变成调用时填的空。三原语的取舍一句话:模型要"做"的事进 tool,Host 要"读"的数据进 resource,人要"填"的模板进 prompt。
5、SQLite 只读查询 Server
进入实战:给 Agent 一个查 SQLite 库的能力,但不给它写库的机会。三个工具不是随手拆的——list_tables → describe_table → query 构成一条"先探索再行动"的路径:模型先看库里有什么表,再看列名与类型,最后才拼 SQL,每一步的输出都是下一步的输入。工具粒度怎么切是第 22 篇的正题,这里先感受这种小工具组合的威力。分工明确后先看前两个:
import sqlite3
from contextlib import closing
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
mcp = MCPServer("sqlite-readonly")
def _connect() -> sqlite3.Connection:
# mode=ro:连接层物理只读,任何写操作直接报错
conn = sqlite3.connect("file:blog.db?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
return conn
@mcp.tool()
def list_tables() -> list[str]:
"""列出库中全部数据表名。"""
sql = "SELECT name FROM sqlite_master WHERE type='table'"
with closing(_connect()) as conn:
return [r["name"] for r in conn.execute(sql)]
@mcp.tool()
def describe_table(table: str) -> list[dict]:
"""查看指定表的列名、类型与默认值。"""
if not table.replace("_", "").isalnum():
raise ToolError(f"非法表名:{table}")
with closing(_connect()) as conn:
rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
return [dict(r) for r in rows]
解读两处设计。连接用 URI 形式 file:blog.db?mode=ro 打开,物理层面锁死只读;describe_table 的表名要拼进 SQL,所以先用 isalnum() 白名单校验——凡拼接进 SQL 的输入必须先过校验,这是第 22 篇护栏设计的主题,这里先给最小实现。另外 raise ToolError 不会让进程崩溃:SDK 把它转成 isError 的工具结果,模型看到错误信息后会自行修正重试。
query 工具是护栏的重头,接在同一文件里:
import re
ALLOWED = re.compile(r"(?is)^\s*(select|with)\b")
BANNED = re.compile(r"(?i)\b(insert|update|delete|drop|alter|create)\b")
@mcp.tool()
def query(sql: str, limit: int = 50) -> list[dict]:
"""只读查询:仅接受单条 SELECT/WITH,结果强制 LIMIT。"""
sql = sql.strip().rstrip(";")
if ";" in sql:
raise ToolError("拒绝多语句执行")
if not ALLOWED.match(sql):
raise ToolError("仅允许 SELECT 或 WITH 开头的语句")
if BANNED.search(sql):
raise ToolError("检测到写操作关键字")
if not re.search(r"(?i)\blimit\s+\d+\s*$", sql):
sql += f" LIMIT {min(max(limit, 1), 200)}" # 上限钳在 200
with closing(_connect()) as conn:
return [dict(r) for r in conn.execute(sql).fetchall()]
if __name__ == "__main__":
mcp.run()
护栏一共四层:剥掉结尾分号后再查分号,堵多语句注入;开头白名单只放行 SELECT 与 WITH(后者是 CTE 语法,合法查询需要它);正文黑名单拦下藏在子查询里的写动词;没有 LIMIT 的查询自动补一个,且上限钳在 200,防止模型一句"查全表"把上下文撑爆。
这个 Server 的安全三件套:连接层
mode=ro物理只读,语句层白名单加黑名单,结果层强制 LIMIT。核心前提只有一条——LLM 生成的 SQL 永远不该被默认信任。
6、两种部署形态
同一个 server.py,两条发布路径:自己机器上用,配 stdio;给团队或线上 Agent 用,挂 Streamable HTTP。
6.1、stdio 本地接入
Claude Desktop 的服务器配置是 JSON 文件:Windows 在 %APPDATA%\Claude\claude_desktop_config.json,macOS 在 ~/Library/Application Support/Claude/claude_desktop_config.json。把刚写的 Server 注册进去:
{
"mcpServers": {
"sqlite-readonly": {
"command": "uv",
"args": ["--directory", "E:/mcp/sqlite-readonly", "run", "server.py"]
}
}
}
解读:mcpServers 下每个键是一个 Server,command 加 args 就是 Host 拉起子进程的完整命令行;保存后重启 Claude Desktop,工具列表里就会出现三个查询工具。懒人路径是 uv run mcp install server.py,由 CLI 替你写这段 JSON。字段细节见 Claude Desktop 本地接入文档。
6.2、HTTP 远程部署
跨机器访问就要换 HTTP。新建 http_server.py,把同一个 mcp 实例包成 ASGI 应用:
from server import mcp # 复用上一节的 Server 实例
import uvicorn
# 传输参数从 v2 起在这里传,不再进构造函数
app = mcp.streamable_http_app(json_response=True)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=9000)
解读三点。其一,streamable_http_app() 返回 Starlette 应用,默认把 MCP 端点挂在 /mcp 路径,既可以让 uvicorn 直接跑,也可以 mount 进现有的 FastAPI 服务(宿主应用的 lifespan 里记得启动 mcp.session_manager.run())。其二,v2 把 host、port、json_response、stateless_http 这些传输参数从构造函数挪到了 run() 与 app 工厂方法——老代码里 MCPServer("x", port=9000) 现在直接 TypeError。其三,不想手写 uvicorn 时,一行 mcp.run(transport="streamable-http", host="0.0.0.0", port=9000) 效果等价。
版本背景:v2 实现的是 2026-07-28 版 MCP 规范,砍掉了会话握手、改为无状态请求——每条 HTTP 请求自包含,过网关、上 Serverless 都不再需要粘性会话,这给远程部署省掉了大量心智负担。
代码里传的 json_response=True 也值得多说一句:它让响应退化为普通 JSON 而不是 SSE 流,牺牲服务端推送能力换取最大兼容性——不少企业内网网关会缓冲甚至掐断流式响应,这种环境下它是保命参数;反过来说,需要进度通知、长任务流的场景就别开它。
7、测试与调试
工具写完别急着接 Claude Desktop,先用官方调试工具 MCP Inspector 过一遍:
uv run mcp dev server.py
# 或用 npx 直接拉起,不挑 Server 语言
npx @modelcontextprotocol/inspector@latest python server.py
两种方式等价,都会拉起 Inspector 并打开浏览器界面:第一条是 mcp CLI 内置的快捷方式,第二条用 npx 现场下载,Node 写的 Server 也一样能测。界面里的开发动线:
- 左侧选传输类型(stdio 或 HTTP),点 Connect
- Tools 页列出全部工具与自动生成的 schema
- 选工具、填参数、点 Call,下方看 content 与 structured_content
- Resources 与 Prompts 页同理可浏览
日志面板能看到 Server 的 stderr——stdio 模式下 Server 崩溃、print 污染 stdout 这类问题在这里一眼定位。先在 Inspector 里把三个工具都调通,再去接 Claude Desktop,是排查成本最低的顺序。Inspector 还提供命令行模式,能列工具清单、发调用请求并断言结果,适合把"Server 还活着"做成 CI 里的冒烟测试。更多用法见 MCP Inspector 文档。
8、总结
这一篇把第 19 篇的协议落成了代码:官方 mcp 包与 v2 的 MCPServer 类名(别装成社区 fastmcp 包);@mcp.tool() 三行注册一个工具,与第 5 篇的 @tool 完全同构;dict 返回自动携带 structured_content;Resources 与 Prompts 补齐数据面与模板面;SQLite 实战的三层只读护栏;stdio 配 Claude Desktop 与 streamable_http_app 挂 HTTP 两种形态;Inspector 先行调试。最值得带走的判断:MCP Server 的代码量几乎全在业务逻辑与护栏上,协议本身已被 SDK 吃掉——写好工具的功夫,从来都在工具之外。
练习:把第 5 篇的汇率工具改造成 MCP Server。新建 MCPServer("exchange") 实例;给 get_exchange_rate 加上 @mcp.tool(),API Key 从环境变量 os.environ 读取,不要写进代码;用 uv run mcp dev 调通任意两种货币互查;最后 mcp install 注册进 Claude Desktop,问它"100 美元能换多少日元",看着它调用你的工具。有余力的,再加一个 resource 暴露支持的币种列表。
对于正在搭 Agent 的同学,建议先把这篇的 Server 接进 Claude Desktop 跑两天,直观感受 Host 与 Server 的分工边界,再进第 21 篇——用 LangChain 的 MCP 适配器把这些 Server 接进你自己的 Agent 图。


