主题
实战:把一个查询能力做成 MCP Server
公司里有一类能力反复被问、但只在内部系统里:商品归类编码(HS)查询、订单状态、库存、客户档案、政策条款。人去查一次要打开系统、填条件、复制结果;AI 想查,却根本没有入口——模型不知道现在几点,也碰不到内网数据库。
三种补救方式:
- 把数据导出给模型——合规和时效都不可接受;
- 给每个 AI 客户端各写一套集成——M 个客户端 × N 个数据源就是 M×N 份定制代码;
- 把它写成一个 MCP Server——一次写好,任何支持 MCP 的客户端都能连。
本文走第三条路。最终交付物是:一个内部查询 Server,本地 stdio 和内网 HTTP 两个端点共用同一份代码,任何同事在客户端里配一条配置就能用,不用你写前端,也不用做 App。
1. 先判断:这个能力值不值得做成 Server
MCP(Model Context Protocol,模型上下文协议)是 AI 应用和外部数据/工具之间的标准插口。它不是产品,也不是模型,而是一套开放协议:怎么发现能力、怎么调用、返回什么格式,都由协议约定。
1.1 从 M×N 到 M+N
没有标准插口时,每一对"客户端 × 数据源"都是一份私有格式的定制集成:
text
M 个客户端 × N 个数据源 = M×N 份定制集成
客户端 A ──┬── 数据源 1(私有格式 A1)
├── 数据源 2(私有格式 A2)
└── 数据源 3(私有格式 A3)
客户端 B ──┬── 数据源 1(私有格式 B1)
├── ...(再来三份)
客户端 C ──┬── ...(再来三份)有了 MCP,客户端和数据源各自只对接协议一次:
text
M + N
客户端 A ─┐
客户端 B ─┼── MCP(一套开放协议)──┬── 数据源 1
客户端 C ─┘ ├── 数据源 2
└── 数据源 3
你写一次 Server,所有支持 MCP 的客户端都能连。1.2 三问判断法
| 要问的问题 | 值得做成 Server | 不值得 |
|---|---|---|
| 会被反复使用吗 | 每天/每周被查几十次 | 一次性脚本,跑完就丢 |
| 会被几个入口访问 | 多个客户端、多个同事、还可能进工作流 | 只有你自己一个客户端、只用一次 |
| 数据能直接给模型吗 | 不能:在内网/数据库里,有权限或合规约束 | 能:直接放进提示词或文件更简单 |
三个问题全是左列,才值得动手。这也是"全公司关务都要查的编码库"比"我这次要算的一个数"更适合做成 Server 的原因。
1.3 什么时候不要做 MCP
- 只用一次:直接写个 Python 函数调 API,五分钟解决,不要为了协议而协议;
- 只有一个消费方:一个客户端、一个人用,配置成本高于收益;
- 数据本来就是公开文本:把内容放进提示词、文件或知识库即可;
- 你要的是"把结果发给我"而不是"让模型随时查":那是定时任务或推送,见《第一个自动化工作流》。
1.4 一个常见误解:MCP 本身不含 AI
MCP 只是协议,就像 HTTP 之于网页——HTTP 里没有网页内容,MCP 里也没有模型。写 MCP Server 就是普通的后端开发:接收请求、查数据、返回 JSON,只是请求格式由协议规定,调用方从人换成 AI 模型。你不需要训练模型,也不需要懂反向传播;你需要懂的是接口设计、权限和数据口径。
2. 定义这件工作
动手前先把任务写清楚,后面的每一步都由这张表约束:
| 项目 | 内容 |
|---|---|
| 场景 | 关务每天查询商品归类编码,数据在内网数据库,AI 客户端碰不到 |
| 输入 | 完整 HS 编码(8/10 位数字)或商品关键词 |
| 输出 | 结构化结果:编码、品名、税率、监管条件、税则版本、来源表 |
| 下游 | 关务人员在现有流程里复核后使用 |
| 风险 | 错归类直接变成申报风险;结果必须可追溯到版本与来源 |
| 验收 | 三个客户端各调通一次;抽样 10 条与源系统一致;越权数据读不到 |
| 不交给模型 | 最终用哪个编码、是否据此申报——判定权和责任归关务 |
从内部能力到多客户端复用
- 1确认这个能力值得复用(三问判断法)人负责复用频率、客户端数量、是否允许直连数据
- 2定义输入、输出、风险与验收人负责这一步不做,后面全是返工
- 3写 Server 骨架:工具、资源、提示词AI 执行可委托 AI 生成第一版
- 4把业务逻辑放进可独立测试的服务层AI 执行协议层不写 SQL
- 5Inspector 验证三种能力检查点Tools / Resources / Prompts 各验一次
- 6接入本地 stdio 与远程 HTTP 两个端点自动触发同一份代码,只换 run() 的 transport
- 7抽样比对结果、检查权限与审计人负责AI 不能自证正确
3. MCP 的三种能力
协议约定了三种能力,区别不在于它们长什么样,而在于谁决定使用:
| 能力 | 谁决定使用 | 相当于 | 本文示例 |
|---|---|---|---|
| Tools(工具) | 模型自己挑、自己调 | 给模型一双手 | get_current_time、lookup_hscode |
| Resources(资源) | 应用/客户端决定读取 | 给模型一双眼睛 | info://server 能力清单与数据口径 |
| Prompts(提示词) | 人从菜单里选(斜杠命令) | 给模型一套标准话术 | time_report、hscode_check |
三者不是"全都写上更好"。写工具就够了的能力,不要额外堆资源和提示词——工具越多,模型选错的几率越高(见《Agent 工具设计与 MCP》)。
4. 十五分钟跑通:time-info Server
先用一个人畜无害的能力走通全流程:查询当前时间。模型不知道现在几点,这是最典型的"补模型短板"。
4.1 安装
bash
uv add "mcp[cli]" # 或 pip install "mcp[cli]"cli 额外包提供 mcp dev / mcp run / mcp install 三个命令。Python 要求 3.10+。
4.2 一个文件,三种能力
python
# server.py —— time-info:一个能力,两种端点
import json
import os
from datetime import datetime, timezone
from typing import Annotated, Literal
from zoneinfo import ZoneInfo
from pydantic import Field
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
mcp = MCPServer("time-info")
@mcp.tool(
title="查询当前时间",
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def get_current_time(
tz: Annotated[
Literal["Asia/Shanghai", "UTC", "America/New_York"],
Field(description="IANA 时区名,只接受这三个值"),
] = "Asia/Shanghai",
) -> dict:
"""查询指定时区的当前时间。问“现在几点”“截止时间是今天吗”时用它。
不要用它做日期推算或历史时间查询——它只回答“此刻”。"""
now = datetime.now(ZoneInfo(tz))
return {
"timezone": tz,
"local_time": now.isoformat(timespec="seconds"),
"utc_offset": now.strftime("%z"),
"weekday": now.strftime("%A"),
"source": "运行 Server 的机器系统时钟",
"fetched_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
}
@mcp.resource("info://server", mime_type="application/json")
def server_info() -> str:
"""这个 Server 的能力清单、时区白名单和数据口径。客户端可随时读取。"""
return json.dumps(
{
"name": "time-info",
"capabilities": {
"tools": ["get_current_time"],
"resources": ["info://server"],
"prompts": ["time_report"],
},
"timezone_allowlist": ["Asia/Shanghai", "UTC", "America/New_York"],
"data_source": "运行 Server 的机器系统时钟,未对时到权威时间源",
},
ensure_ascii=False,
indent=2,
)
@mcp.prompt(title="跨时区时间核对")
def time_report(tz: str = "Asia/Shanghai") -> str:
"""把“现在几点”变成一句可核对的标准话术。由使用者从菜单里选,不是模型自动调用。"""
return (
f"请调用 get_current_time 工具查询时区 {tz} 的当前时间,然后用一句话汇报:"
"时区、当地时间、UTC 偏移、星期几,并注明数据来自 Server 机器时钟。"
)
if __name__ == "__main__":
if os.getenv("MCP_TRANSPORT") == "http":
mcp.run(
transport="streamable-http",
host=os.getenv("MCP_HOST", "127.0.0.1"),
port=int(os.getenv("MCP_PORT", "8000")),
)
else:
mcp.run()注意你没有写的东西:没有 JSON Schema(tz: ... 的类型注解就是 Schema)、没有请求解析、没有参数校验代码。SDK 从函数名、docstring 和类型注解生成协议需要的一切——docstring 是模型看到的工具描述,写清楚"何时用、何时不用"比什么都重要。
4.3 用 Inspector 验证
bash
uv run mcp dev server.pyInspector 做的事和真实客户端完全一样:把 server.py 当子进程启动,用 stdio 连上去。依次打开三个页签各验一次:
- Tools:调用
get_current_time,传一个非法时区(如Mars/Olympus),看是否被 Schema 挡住并返回模型可读的错误; - Resources:读
info://server,确认口径说明完整; - Prompts:渲染
time_report,确认生成的用户消息是你想要的话术。
stdio 下 stdout 是协议通道。调试不要用
print(),用logging(它写到 stderr)。否则会污染协议流,表现为客户端"莫名其妙连不上"。
4.4 接进本地客户端
客户端配置文件(Claude Code、Cursor、VS Code 等结构一致):
json
{
"mcpServers": {
"time-info": {
"command": "uv",
"args": ["run", "--directory", "/abs/path/to/time-info", "python", "server.py"]
}
}
}重启客户端,问一句"现在上海几点",看到它调用 get_current_time,本地端点就通了。
5. 同一份代码,第二个端点:Streamable HTTP
5.1 换一行 run()
你已经写好了——第 4.2 节末尾的 MCP_TRANSPORT=http 分支就是远程端点:
bash
MCP_TRANSPORT=http MCP_HOST=127.0.0.1 MCP_PORT=8000 python server.py
# 端点:http://127.0.0.1:8000/mcp官方 CLI 也能做同样的事:uv run mcp run server.py --transport streamable-http。客户端配置从"启动命令"改成"一个 URL":
json
{
"mcpServers": {
"time-info-remote": {
"type": "http",
"url": "https://mcp.internal.example.com/mcp"
}
}
}这就是第 1.1 节的 M+N:同一份业务代码,本地 stdio 和远程 HTTP 两种接法,同事只要一个 URL,不用装 Python、不用拿代码。
5.2 部署到内网要注意的三件事
host默认127.0.0.1,只监听本机。改成0.0.0.0意味着任何能访问该端口的人都能调用你的工具——必须同时上鉴权(反向代理、网关或内网隔离),不能裸奔;- 多副本部署用
stateless_http=True,每个请求一个独立传输,不依赖进程内会话状态; - 更远的部署问题(DNS rebinding 防护、HTTPS 终止、超时)见官方 SDK 的部署章节,不要凭记忆猜参数。
凭证一律走环境变量或密钥管理,绝不写进配置文件和仓库,详见《AI 使用安全清单》。
6. 换成真实业务:关务编码查询 Server
现在把同样的骨架换成真业务。差别只在三处:分层、工具设计、权限边界。
6.1 分层:协议层不写 SQL
text
hscode-mcp/
├── server.py # 只做协议适配:装饰器、Schema、结果映射
├── service.py # 业务规则:鉴权、校验、口径、审计
├── repository.py # 只读数据库访问
└── tests/
└── test_tools.pyMCP 只是一个适配器。业务逻辑放在能独立测试的服务层,将来加 REST 接口或队列消费者时,规则不用重写一遍。
6.2 工具设计:窄而明确
python
# server.py(节选)
from typing import Annotated
from pydantic import Field
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
from service import lookup_code, search_candidates
mcp = MCPServer("hscode")
@mcp.tool(
title="按编码查归类",
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def lookup_hscode(
code: Annotated[str, Field(pattern=r"^\d{8,10}$", description="HS 编码,8 或 10 位数字")],
) -> dict:
"""按完整 HS 编码查询品名、税率、监管条件和版本日期。
仅在编码已确定时使用;编码未定时先用 search_hscode 找候选。
返回含 version 与 source,使用时必须一并展示以便人工复核。"""
return lookup_code(code)
@mcp.tool(
title="按关键词找候选编码",
annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def search_hscode(
keyword: Annotated[str, Field(min_length=2, max_length=40, description="商品关键词或品名片段")],
limit: Annotated[int, Field(ge=1, le=20)] = 5,
) -> dict:
"""按关键词检索候选 HS 编码,只返回编码与品名,不返回税率。
最终使用哪个编码由关务判断,不要用它直接定稿申报。"""
return search_candidates(keyword, limit)两个刻意的设计:检索工具不返回税率(逼模型走精确查询拿权威值),不提供"帮我确定用哪个编码"的工具(判定权归人)。工具边界就是责任边界——这一点在《Agent 工具设计与 MCP》里有更完整的展开。
6.3 返回必须带证据
python
# service.py(节选)
def lookup_code(code: str) -> dict:
row = repository.fetch_by_code(code) # 只读账号
if row is None:
return {
"found": False,
"code": code,
"next_actions": ["search_hscode", "ask_customs_specialist"],
}
return {
"found": True,
"code": row.code,
"product_name": row.product_name,
"vat_rate": row.vat_rate,
"supervision_conditions": row.conditions,
"version": row.version, # 税则版本
"effective_from": row.effective_from,
"source": row.source, # 内部库表或文件编号
"disclaimer": "编码适用性由关务复核确认,AI 结果不作为申报依据",
}found=False 时给出 next_actions——模型拿到的是"下一步该做什么",而不是 Something went wrong。错误要能被模型读懂并自我纠正。
6.4 权限边界:数据留在 Server 侧
MCP 的价值之一就是不让模型直连数据库:AI 只能通过你暴露的工具按约定访问。
- 只读账号:连接串从环境变量读,账号只给 SELECT,禁止写操作;
- 返回字段最小化:不返回客户名、价格、供应商等无关字段;
- 调用级审计:记录谁、何时、查了哪个编码;
- 判定权归人:高风险结论必须人工复核,工具不替人拍板。
协议统一不等于自动安全。鉴权、租户隔离、输入校验、限流和审计,仍然要你自己实现。
6.5 让 AI 写第一版:可复制的委托 Prompt
text
背景:你是公司内部的 Python 后端开发。我们有一个商品归类编码(HS)查询库,
现在要让多个 AI 客户端(IDE 插件、桌面客户端、内网页)都能用自然语言查它。
目标:产出 MCP Server 第一版,本地 stdio 与内网 HTTP 两种端点共用同一份代码。
输入:
- 数据库表结构(见下方 DDL)
- 现有查询函数(见 repository.py)
- 团队规范:只读账号、返回字段最小化、每次调用写审计日志
任务:
1. 用 Python MCP SDK v2(from mcp.server import MCPServer)实现 server.py,
暴露两个工具 lookup_hscode / search_hscode,一个资源 info://server
2. 业务逻辑放在 service.py,server.py 只做协议适配,不写 SQL
3. 每个工具返回结构化 dict,必须含 version 与 source 字段
4. 支持环境变量 MCP_TRANSPORT=http 切换到 streamable-http
5. 写 tests/test_tools.py,用 SDK 的 in-memory Client 验证工具与参数校验
约束:
- 只用只读账号,连接串从环境变量读取,不写进代码和文档
- 不实现任何“自动确定编码”的工具,判定权归关务
- 不输出客户、价格、供应商等无关字段
- 不要在 Server 代码里 print 调试(stdio 下 stdout 是协议通道)
输出格式:按文件列出完整代码,每个文件前写一段三行的职责说明
验收标准:
- mcp dev server.py 能在 Inspector 里看到 2 个工具、1 个资源
- 传入非法编码(如 "abc")时返回模型可读的错误,而不是崩溃
- 每个查询结果能指出它来自哪个版本和来源表
- 全部查询走只读账号,代码中没有写操作7. 人检查什么
AI 能写完代码、能说"已通过测试",但正确性判断在你:
7.1 事实
- 随机抽 10 条编码,把 Server 返回与源系统逐字段比对(税率、监管条件、版本日期);
- 确认
version与effective_from是真实值,不是占位符; - 边界样本:停用编码、新启用编码、10 位子目。
7.2 风险
- 代码里确实没有写操作,数据库账号是只读;
- 返回结果不含客户、价格、供应商;
- 审计日志有输出,
host不是裸奔的0.0.0.0; - 敏感凭证没有进配置文件和 git。
7.3 完整性
- 查不到时有
found=False与next_actions,不是抛异常; - 超时和数据库不可用时返回可分类的错误,而不是挂住;
- 两个端点(stdio、HTTP)各用真实客户端调通一次。
8. 验收清单
- [ ] 三问判断法确认这个能力值得做成 Server,而不是一次性脚本;
- [ ] 输入、输出、风险、验收方法写成了文档里的表;
- [ ] 工具描述写清了"何时用、何时不用",参数有枚举、范围或格式约束;
- [ ] 协议层不写 SQL,业务规则在可独立测试的服务层;
- [ ] 返回结果带版本与来源,查不到时给
next_actions; - [ ] 只用只读账号,连接串走环境变量,无写操作;
- [ ] 返回字段最小化,不含无关的客户、价格、供应商信息;
- [ ] 有调用级审计日志(谁、何时、查了什么);
- [ ] Inspector 里 Tools / Resources / Prompts 各验一次;
- [ ] 本地 stdio 与远程 HTTP 两个端点各用真实客户端调通一次;
- [ ] 抽样 10 条结果与源系统一致;
- [ ] 高风险判定权明确留在人手里,工具不替人拍板。
9. 做完一次之后
9.1 保存模板
把第 4.2 节的骨架存成团队模板:一个 Server 对象 + 一个工具 + 一个资源 + 一个提示词 + 双端点 run()。下一个内部能力照抄骨架,只替换 service.py。
9.2 接进工作流
Server 不只服务于聊天窗口。接入《第一个自动化工作流》的情报流或定时任务时,它复用的是同一组工具,不必为每个流程重写集成——这才是 M+N 的复利所在。
9.3 建立回归集
每次改工具描述或 Schema,跑一遍固定的调用样本,记录工具选择准确率和参数通过率。加一个新工具导致旧的选择准确率下降,说明语义重叠,先合并再上线。
10. 版本快照(最后核对 2026-09-29)
协议和 SDK 变化很快,具体版本号不要记在脑子里,用这张表对照一手来源:
| 事项 | 当前状态 | 一手来源 |
|---|---|---|
| 规范版本 | 2026-07-28 稳定版:无状态化,移除 initialize 握手,新增 server/discover,结果带 resultType | Key Changes |
| 传输方式 | stdio(本地子进程,默认)、streamable-http(部署用)、sse(已废弃,勿用于新项目) | Running your server |
| HTTP 端点 | Streamable HTTP 默认路径 /mcp,默认端口 8000 | 同上 |
| Python SDK | v2 稳定线:from mcp.server import MCPServer;v1 的 FastMCP 写法需迁移,pip install mcp 现在装 2.x | python-sdk README |
| 治理 | 2025-12-09 Linux 基金会成立 Agentic AI Foundation,Anthropic 将 MCP 捐入 | Linux Foundation 新闻稿 |
十分钟自查:看规范站的 changelog 是否换了版本日期 → 看 SDK README 的版本徽标 → 跑 uv run mcp version 确认本地版本 → 用 Inspector 连一次确认传输可用。
11. 下一步
- 想先补原理,看《Agent 与 MCP》的工具调用循环;
- 想把工具设计到生产级,看《Agent 工具设计与 MCP》;
- 想在客户端里配置现成 Server,看《MCP 实战指南》;
- 想把它接进流程,看《从工作流到 Agent》和《第一个自动化工作流》。
一句话收尾:MCP 是让 AI 够得到你的系统的那根标准插口。 值得反复被访问的能力,才配做成 Server——做完第一个,第二个只是换一个 service.py。