AI Agent 开发(20):MCP Server 实战

QuibblerAgentQuibblerAgent 2026-10-25 约 25 分钟 3 次阅读

第 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 也一样能测。界面里的开发动线:

  1. 左侧选传输类型(stdio 或 HTTP),点 Connect
  2. Tools 页列出全部工具与自动生成的 schema
  3. 选工具、填参数、点 Call,下方看 content 与 structured_content
  4. 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 图。

相关推荐

Hermes Agent 详解:自进化学习型 AI 代理
Agent

Hermes Agent 详解:自进化学习型 AI 代理

Hermes Agent 详解:自进化学习型 AI 代理Hermes Agent 是由 Nous Research 于 2026 年 2 月开源的自主 AI 代理框架,其核心定位是"与你共同成长的智能体"。与传统的 IDE 编程助手或单一 API 聊天机器人不同,Hermes 运行在你的服务器上,能够记住所学内容,运行时间越长能力越强。它采用 MIT 开源许可,支持 Linux、macOS 和 W

393
Hermes Agent:与你一起成长的开源 AI 代理
Agent

Hermes Agent:与你一起成长的开源 AI 代理

Hermes Agent:与你一起成长的开源 AI 代理大多数 AI 助手是"健忘的租客"——换个窗口就失忆、换个平台就重学、干过的活不留下任何沉淀。Nous Research 出品的 Hermes Agent 反其道而行,口号是"The Agent That Grows With You"(与你一起成长的代理):它是一个开源(MIT)、自托管的 AI 代理,具备持久记忆(学习你的项目、自动生成技

322
speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线
AI

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线语音 Agent 的"全开源"路线一直有个尴尬:框架开源了,模型却离不开云端 API。huggingface/speech-to-speech 把这条路走通了:VAD → STT → LLM → TTS 四段低延迟流水线,每个组件都可换成开源模型本地跑——Apple Silicon 一台 Mac 就能撑起

264