上个月帮朋友看一家小电商的订单数据。库在本地，一个几百兆的 SQLite 文件，我这边有 Claude，那边有数据，中间隔着一道复制粘贴的墙——想让它按月统计营业额，我得先写 SQL、导出 CSV、再贴进对话框。来回几次之后我放弃了，转头去写了个 MCP Server，前后不到两百行代码。这篇就把整个过程记下来，包括中间被卡住的几个地方。
MCP 解决的其实是”接什么”的问题
在 MCP 出现之前，想让模型读你本地的东西，路数无非几种：写个函数调用的 JSON schema 塞进 system prompt，或者干脆自己搞一套工具注册。问题是每换一个客户端（Claude Desktop、Cursor、你自己写的 Agent）就得重新适配一遍，工具定义、参数格式、错误处理全都要重写。
MCP 把这一层定死了。它本质上就是一套基于 JSON-RPC 的协议，客户端负责启动你的进程，双方通过 stdin/stdout 交换消息。你只要按照协议的约定暴露能力，任何支持 MCP 的客户端都能直接用。
协议里定义了三种原语，初学的时候容易混：
- tool：模型主动调用的函数，可以带参数，可能有副作用。查询、下单、发邮件都走这个。
- resource：模型读取的上下文，用 URI 寻址，类似 HTTP 的 GET，理想情况下是幂等的。
- prompt：预置的提示词模板，一般由用户在界面上手动触发，不是模型自己调。
我的划分标准很简单：需要参数、可能改变状态，就做成 tool；只是”读一份数据喂给模型”，做成 resource；需要引导模型按固定套路做事的，做成 prompt。
环境准备
需要 Python 3.10 以上，SDK 是官方那个 mcp 包。用 pip 或者 uv 都行：
pip install "mcp[cli]"
# 或者
uv add mcp后面为了演示方便，我用的都是标准库里的 sqlite3，不用额外装东西。
先造点测试数据
建一个 init_db.py，生成一张订单表灌点假数据。别跳过这步，用真实点的数据调工具，比对着空表发呆有用得多。
import sqlite3
from pathlib import Path
DB = Path(__file__).with_name("shop.db")
rows = [
(1, "杭州云栖科技", 12800.0, "paid", "2025-03-02T09:14:00"),
(2, "成都锦鲤工作室", 4300.5, "pending", "2025-03-03T11:02:00"),
(3, "深圳前海数据", 27600.0, "paid", "2025-03-05T16:40:00"),
(4, "杭州云栖科技", 980.0, "refunded", "2025-03-07T08:21:00"),
(5, "北京青梧网络", 15200.0, "paid", "2025-03-09T13:55:00"),
(6, "成都锦鲤工作室", 7600.0, "pending", "2025-03-11T10:08:00"),
(7, "上海望川信息", 3320.0, "paid", "2025-03-12T15:30:00"),
(8, "深圳前海数据", 45900.0, "paid", "2025-03-14T09:47:00"),
]
with sqlite3.connect(DB) as conn:
conn.execute("DROP TABLE IF EXISTS orders")
conn.execute("""
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
customer TEXT NOT NULL,
amount REAL NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL
)
""")
conn.executemany(
"INSERT INTO orders (id, customer, amount, status, created_at) "
"VALUES (?, ?, ?, ?, ?)",
rows,
)
conn.execute("CREATE INDEX idx_orders_status ON orders(status)")
conn.commit()
print(f"已写入 {len(rows)} 条订单 -> {DB}")跑一下：python init_db.py。同目录下会多出一个 shop.db。
先跑通一个 Hello World
别一上来就写业务逻辑。先用最小的例子确认环境没问题：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo")
@mcp.tool()
def add(a: int, b: int) -> int:
"""计算两个整数之和。"""
return a + b
if __name__ == "__main__":
mcp.run()直接运行这个脚本，它不会打印任何东西，而是挂在 stdio 上等 JSON-RPC 消息。你可以手动往终端里敲一行 {"jsonrpc":"2.0","id":1,"method":"initialize","params":{...}}，能看到回应就说明通了。不过手敲太麻烦，我们后面用官方客户端来验证。
正式版：把订单库包进来
下面是完整的 server.py，我拆成几段讲。
连接层：只读打开，顺带把关闭问题解决掉
from __future__ import annotations
import json
import logging
import re
import sqlite3
import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Any
from mcp.server.fastmcp import Context, FastMCP
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("shop-mcp")
DB_PATH = Path(__file__).with_name("shop.db")
mcp = FastMCP("shop-sqlite")
_READ_ONLY = re.compile(r"^s*(select|with)b", re.IGNORECASE)
@contextmanager
def _connect():
"""以只读模式打开数据库。
mode=ro 由 SQLite 自己兜底，任何写入都会直接抛 OperationalError，
比在应用层用正则过滤靠谱得多。
"""
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=5)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()这里有两个点值得说。第一，mode=ro 是真正的安全边界，不在应用层做字符串判断。第二，sqlite3.Connection 的上下文管理器只管事务提交/回滚，不会关闭连接。我之前直接写 with sqlite3.connect(...) as conn，跑了几十次之后进程里堆了一堆没关的句柄。用 contextmanager 自己包一层才干净。
工具一：看看有哪些表
@mcp.tool()
def list_tables() -> list[str]:
"""列出当前数据库里所有的表名。返回的名字可以直接传给 describe_table。"""
with _connect() as conn:
rows = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name NOT LIKE 'sqlite_%' "
"ORDER BY name"
).fetchall()
return [r["name"] for r in rows]工具二：看看表结构
@mcp.tool()
def describe_table(table: str) -> dict[str, Any]:
"""查看某张表的字段结构、字段类型和总行数。
Args:
table: 表名，必须是 list_tables 返回过的名字之一。
"""
if table not in list_tables():
raise ValueError(f"表 {table!r} 不存在，请先调用 list_tables 确认")
# PRAGMA 不支持参数占位符，这里的 f-string 是安全的，
# 因为 table 刚刚跟白名单比对过。
with _connect() as conn:
cols = conn.execute(f"PRAGMA table_info({table})").fetchall()
total = conn.execute(f"SELECT COUNT(*) AS n FROM {table}").fetchone()["n"]
return {
"table": table,
"row_count": total,
"columns": [
{"name": c["name"], "type": c["type"], "notnull": bool(c["notnull"])}
for c in cols
],
}工具三：执行查询
@mcp.tool()
def run_query(sql: str, limit: int = 50) -> dict[str, Any]:
"""对订单库执行一条只读 SQL 查询。
只接受 SELECT 或 WITH 开头的语句，写操作会被数据库直接拒绝。
返回行数超过 limit 时只返回前 limit 行，并把 truncated 置为 true。
Args:
sql: 单条 SQL 语句，不要带结尾分号。
limit: 最多返回多少行，默认 50，上限 500。
"""
cleaned = sql.strip().rstrip(";")
if not _READ_ONLY.match(cleaned):
raise ValueError("只允许 SELECT / WITH 查询，请重写语句")
limit = max(1, min(limit, 500))
with _connect() as conn:
cursor = conn.execute(cleaned)
fetched = cursor.fetchmany(limit + 1)
columns = [d[0] for d in cursor.description] if cursor.description else []
records = [dict(r) for r in fetched[:limit]]
return {
"columns": columns,
"rows": records,
"truncated": len(fetched) > limit,
}多取一行用来判断是否被截断，是个小技巧，免得模型拿到 50 行数据以为这就是全部。
资源：把 schema 直接喂过去
模型写 SQL 之前得知道字段含义。让它在每次查询前都调一遍 describe_table 不是不行，但多一轮往返。更省事的做法是把建表语句做成资源：
@mcp.resource("schema://orders")
def orders_schema() -> str:
"""orders 表的建表语句和字段含义，写 SQL 前可以先读这个。"""
return (
"CREATE TABLE orders (n"
" id INTEGER PRIMARY KEY, -- 订单号n"
" customer TEXT NOT NULL, -- 客户名称n"
" amount REAL NOT NULL, -- 金额，单位：元n"
" status TEXT NOT NULL, -- 取值：paid / pending / refundedn"
" created_at TEXT NOT NULL -- ISO8601 本地时间，无时区n"
")n"
)资源 URI 里带花括号会自动变成模板，可以传参：
@mcp.resource("orders://{order_id}")
def order_detail(order_id: str) -> str:
"""按订单号读取单条订单的完整信息。"""
with _connect() as conn:
row = conn.execute(
"SELECT * FROM orders WHERE id = ?", (order_id,)
).fetchone()
if row is None:
return f"订单 {order_id} 不存在"
return json.dumps(dict(row), ensure_ascii=False, indent=2)再加一个提示词模板
@mcp.prompt()
def revenue_report(month: str) -> str:
"""生成按月分析营收的提示词。
Args:
month: 形如 2025-03 的月份字符串。
"""
return (
f"请分析 {month} 的订单数据。先用 run_query 查出该月每天的成交额，"
"再指出波动最大的三天，最后给出三条可执行的建议。"
"所有金额以人民币元为单位，不要编造数据。"
)
if __name__ == "__main__":
mcp.run()工具描述到底该写多细
这是最容易糊弄、也最影响效果的一环。函数的 docstring 会被直接抽成工具的 description，模型就是靠它决定要不要调、怎么填参数。
对比一下两种写法。差的是这样：
@mcp.tool()
def run_query(sql: str) -> list:
"""查询数据库。"""模型看到这个，不知道能查什么表、会不会写坏数据、返回值长什么样，于是要么不敢调，要么乱调。好的写法里至少要说清三件事：能做什么、有什么限制、参数怎么填。我上面那个 run_query 的描述就在做这件事——明确告诉它只支持 SELECT、返回会被截断、参数带上限。
还有个细节：返回值类型别偷懒写 Any 或者自定义类。SDK 会把它解析成 JSON Schema 发给客户端，写 dict[str, Any] 或者 list[str] 这种具体结构，模型更容易理解返回了什么。
接进 Claude Desktop
配置文件的位置：macOS 在 ~/Library/Application Support/Claude/claude_desktop_config.json，Windows 在 %APPDATA%Claudeclaude_desktop_config.json。没有就新建一个。
{
"mcpServers": {
"shop-sqlite": {
"command": "python",
"args": ["/Users/yourname/code/mcp-shop/server.py"]
}
}
}路径一定写绝对路径。相对路径在客户端启动子进程时的工作目录是不确定的，我因为这个排查了好一会儿。改完配置要完全退出 Claude Desktop 再重启，光关窗口不行。
重启之后对话框左下角会出现一个工具图标，点开能看见 list_tables、describe_table、run_query 三个工具。这时候可以直接问它”3 月份各状态订单的金额分布”，它会自己去调。
几个把我卡住的坑
stdout 是协议通道，一行 print 都不能有。 我在 run_query 里加过一个调试用的 print(sql)，结果客户端直接报连接失败，日志里一堆 JSON 解析错误。所有调试输出走 stderr，我上面配的 logging.basicConfig(stream=sys.stderr) 就是干这个的。
异常信息要写给人看。 FastMCP 会把异常转成错误消息回传给客户端，模型能看到内容。所以 raise ValueError("表 'ordres' 不存在，请先调用 list_tables 确认") 比裸的 KeyError 有用得多——模型能看懂提示，自己纠正拼写。
别在模块顶层打开数据库连接。 客户端会反复重启子进程，模块级的连接会一直占着文件锁。放在函数里按需打开，用完就关。
SQLite 的连接上下文管理器不关连接。 这个前面说过了，很容易踩。
写个客户端脚本做回归测试
每次改完代码都重启 Claude Desktop 去点一遍，效率太低。官方 SDK 自带客户端，写个脚本直接测：
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from pydantic import AnyUrl
SERVER = StdioServerParameters(command="python", args=["server.py"])
async def main():
async with stdio_client(SERVER) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
print("已注册工具:", [t.name for t in tools.tools])
result = await session.call_tool(
"run_query",
{
"sql": (
"SELECT status, COUNT(*) AS cnt, SUM(amount) AS total "
"FROM orders GROUP BY status"
)
},
)
for block in result.content:
print("查询结果:", block.text)
res = await session.read_resource(AnyUrl("orders://3"))
print("订单 3:", res.contents[0].text)
asyncio.run(main())跑一遍，如果工具列表和查询结果都对，说明服务端逻辑没问题，剩下就是客户端配置的事了。
改成远程 HTTP 模式
stdio 模式只能跑在本机，别人用不了。SDK 从 1.8 开始支持 streamable-http，把上面那行稍微改一下就行：
mcp = FastMCP("shop-sqlite", host="127.0.0.1", port=8000)
# ...
if __name__ == "__main__":
mcp.run(transport="streamable-http")端点固定在 http://127.0.0.1:8000/mcp。客户端换成：
from mcp.client.streamable_http import streamablehttp_client
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
# ... 后面和 stdio 版本完全一样注意这个模式本身不带鉴权。要放到内网机器上，前面至少挂个反向代理加 token 校验，别直接暴露到公网。
还能往哪扩
把口子开成万能 SQL 工具，短期省事，长期是隐患——模型写错 SQL 的概率不低，尤其是在字段名有歧义的时候。生产环境我更推荐收敛成几个固定语义的工具，比如 monthly_revenue(month)、top_customers(n)、pending_orders()，SQL 写在服务端，模型只负责选工具和填参数。这样既能保证结果正确，也顺便把权限切得更细。
另外，如果某个查询要跑十几秒，可以用 Context 回报进度，避免客户端显示超时：
@mcp.tool()
async def heavy_report(ctx: Context) -> str:
await ctx.info("开始扫描订单表")
await ctx.report_progress(0.5, 1.0)
return "完成"说白了，MCP Server 就是一个普通的 Python 进程，只不过它说话的方式被协议规定好了。想清楚暴露哪些能力、每个能力的边界在哪，代码本身其实很薄。
