前阵子帮一个做电商运营的朋友处理数据，他的诉求特别朴素：手里有个几十万行的订单 SQLite 文件，想让桌面上的 AI 客户端直接帮他查。以前的做法是把 CSV 导出来，一段一段贴进对话框，贴到第三段就开始丢字段，模型记不住前面的表结构，来回解释比我自己写 SQL 还累。
问题的核心不在于模型不够聪明，而在于模型和我的数据之间缺一根线。它能说会道，但看不见我磁盘上的那个 .db 文件。
MCP（Model Context Protocol）就是这根线。它规定了 AI 客户端和外部工具之间怎么握手、怎么描述能力、怎么传参、怎么返回结果。你写一个符合协议的进程，客户端就能把你的函数当成自己的工具来用。这篇文章不讲概念，直接动手写一个完整的 MCP Server，把本地 SQLite 变成一组可调用的工具，最后挂到客户端里跑通。
先把环境和最小例子跑起来
我习惯用 uv 管理这类小工具，装依赖快，跑脚本也不用激活虚拟环境。如果你用 pip 也完全没问题，把命令换掉就行。
uv init sqlite-mcp
cd sqlite-mcp
uv add "mcp[cli]"
那个 [cli] 别省，它会把官方自带的调试工具 mcp 命令一起装上，后面排查问题全靠它。
先写一个只有加法功能的版本，确认链路是通的。新建 server.py：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("sqlite-explorer")
@mcp.tool()
def add(a: int, b: int) -> int:
"""把两个整数相加。"""
return a + b
if __name__ == "__main__":
mcp.run()
这段代码里有三件事值得注意。
第一，@mcp.tool() 装饰器是全部工作的入口。被它标记的函数会自动变成客户端可见的工具，参数类型注解会被转换成 JSON Schema，模型的调用参数就按这个 schema 生成。
第二，函数名和 docstring 不是给你看的，是给模型看的。模型选不选你这个工具，很大程度上取决于工具名够不够直白、docstring 有没有讲清楚什么时候该用。写 add 比写 do_math 好，写“把两个整数相加”比写“数学函数”好。
第三，mcp.run() 默认走 stdio 传输。客户端会把你的进程当成子进程拉起来，通过标准输入输出收发 JSON-RPC 消息。这是本地场景最省事的模式，不用开端口，不用管鉴权。
跑一下调试器看看效果：
uv run mcp dev server.py
浏览器会自动打开 MCP Inspector，你能在界面上看到 server 暴露的工具列表，手动填参数点一下，结果立刻返回。这个界面后面会反复用到。
接上真实的数据库
加法跑通了，现在换成真家伙。目标是把一个 SQLite 文件暴露出去，让模型能看结构、能查数据。
先造一份测试数据，省得你手头没有现成的库：
import sqlite3, random
from datetime import date, timedelta
conn = sqlite3.connect("demo.db")
conn.execute("""
create table if not exists orders (
order_id text primary key,
customer text not null,
channel text,
amount real,
status text,
created_at text
)
""")
conn.execute("""
create table if not exists customers (
customer text primary key,
city text,
level text,
joined_at text
)
""")
channels = ["app", "h5", "mini_program", "offline"]
statuses = ["paid", "refunded", "pending", "shipped"]
cities = ["杭州", "成都", "广州", "西安", "武汉"]
for i in range(20000):
cid = f"C{random.randint(1000, 1999)}"
conn.execute(
"insert or ignore into customers values (?,?,?,?)",
(cid, random.choice(cities), random.choice(["normal", "vip"]),
str(date(2024, 1, 1) + timedelta(days=random.randint(0, 500))))
)
conn.execute(
"insert into orders values (?,?,?,?,?,?)",
(f"O{i:07d}", cid, random.choice(channels),
round(random.uniform(20, 3000), 2), random.choice(statuses),
str(date(2025, 1, 1) + timedelta(days=random.randint(0, 300))))
)
conn.commit()
conn.close()
两万行订单，够用来验证分页和截断了。
下面是 server 的核心。我把数据库连接抽成了一个只读函数，这一点非常重要，后面会解释原因。
import json
import sqlite3
from pathlib import Path
from mcp.server.fastmcp import FastMCP
DB_PATH = Path(__file__).parent / "demo.db"
mcp = FastMCP("sqlite-explorer")
def _connect() -> sqlite3.Connection:
# mode=ro 让 SQLite 在文件层面拒绝任何写入
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
return conn
用 file:...?mode=ro 打开，而不是普通的 connect()，好处是安全边界落在 SQLite 自己身上。哪怕我的 SQL 校验被绕过了，写操作也会在数据库层抛异常。两层防线总比一层踏实。
一个能看结构的工具
模型第一次接触陌生数据库的时候，最需要知道的是“里面有什么表、各有多少行”。这个信息不给，它就只能瞎猜表名。
@mcp.tool()
def list_tables() -> str:
"""列出数据库中所有用户表，以及每张表的行数。当你不确定数据里有什么的时候，先调用这个。"""
with _connect() as conn:
names = [
row["name"]
for row in conn.execute(
"select name from sqlite_master "
"where type='table' and name not like 'sqlite_%' order by name"
)
]
result = []
for name in names:
count = conn.execute(
f'select count(*) as c from "{name}"'
).fetchone()["c"]
result.append({"table": name, "rows": count})
return json.dumps(result, ensure_ascii=False)
返回 JSON 字符串而不是 Python dict，是因为工具结果最终要被序列化进协议消息。直接返回字符串，字段名和结构完全由我控制，模型也更容易稳定解析。
表名拼接那里用了双引号包裹，这是个容易被忽略的细节。如果表名里带空格或者连字符，不加引号的 SQL 会直接报语法错误。这里的 name 来自 sqlite_master，是我信任的来源，所以拼接是安全的；但如果表名来自模型输入，就必须换成参数化或者白名单校验。
再加一个看字段的工具：
@mcp.tool()
def describe_table(table: str) -> str:
"""查看指定表的建表语句，包含所有字段名和类型。参数 table 是表名，例如 orders。"""
with _connect() as conn:
row = conn.execute(
"select sql from sqlite_master where type='table' and name = ?",
(table,),
).fetchone()
if row is None:
raise ValueError(f"表 {table} 不存在，请先用 list_tables 查看可用表名")
return row["sql"]
注意这里的异常处理方式。我没有 try/except 包一层再返回错误字符串，而是直接抛 ValueError。FastMCP 会把它捕获成一次工具调用失败，把消息原样送回客户端。模型看到“表 xxx 不存在，请先用 list_tables”，下一轮就会自己去调 list_tables 重试。
如果你把异常吞掉、返回一句“查询失败”，模型就会在原地打转，因为它不知道下一步该干什么。错误信息写得好不好，直接决定多轮对话的效率。
查询工具：安全边界放在哪
最危险的就是这个工具，模型生成的 SQL 什么都有可能。我的策略是三道闸门：语句数量、语句类型、关键字黑名单。
_BLOCKED = (
"attach", "detach", "pragma", "insert", "update", "delete",
"drop", "create", "alter", "replace", "vacuum", "reindex",
)
def _guard(sql: str) -> str:
statement = sql.strip().rstrip(";").strip()
if not statement:
raise ValueError("SQL 不能为空")
if ";" in statement:
raise ValueError("一次只能执行一条语句，请去掉分号后的内容")
lowered = statement.lower()
if not lowered.startswith(("select", "with")):
raise ValueError("只允许 SELECT 或 WITH 开头的只读查询")
for word in _BLOCKED:
if word in lowered:
raise ValueError(f"检测到被禁用的关键字：{word}")
return statement
说实话，这种子串匹配的方式并不严谨。如果某张表里刚好有个字段叫 created_at，名字里带 create，就会被误伤。生产环境里应该做词法分析，或者干脆用 SQLite 自己的 set_authorizer 回调，在 SQLite 准备语句的时候逐个判断操作类型，那才是真正可靠的方案。
不过对于个人工具，字符串校验配合 mode=ro 已经足够了：字符串校验负责给出友好的错误提示，mode=ro 负责兜底。
@mcp.tool()
def run_query(sql: str, limit: int = 50) -> str:
"""执行只读 SQL 查询并返回结果。
sql: 要执行的 SELECT / WITH 语句，SQLite 语法。
limit: 最多返回多少行，默认 50，上限 200。
返回 JSON，包含 columns（字段名）、rows（数据）和 truncated（是否被截断）。
"""
statement = _guard(sql)
limit = max(1, min(int(limit), 200))
with _connect() as conn:
cursor = conn.execute(statement)
columns = [d[0] for d in cursor.description]
fetched = cursor.fetchmany(limit + 1)
truncated = len(fetched) > limit
rows = [dict(r) for r in fetched[:limit]]
return json.dumps(
{
"columns": columns,
"rows": rows,
"returned": len(rows),
"truncated": truncated,
},
ensure_ascii=False,
default=str,
)
这里有两个我自己踩过之后才加上的处理。
第一，fetchmany(limit + 1) 然后看长度。多取一行，就能判断结果是不是被截断了，而且不用先跑一次 count。截断这件事必须显式告诉模型，否则它会拿 50 行样本当成全量数据下结论，得出的结论可能是完全反的。
第二，default=str。json.dumps 遇到 datetime、Decimal 这类对象会直接抛 TypeError，而 SQLite 里取出来的东西类型不一定那么干净。default=str 让序列化不至于整个崩掉。
还有个隐藏问题：如果模型写了个 select * from orders，返回 200 行乘 6 列，这个 JSON 轻松就有几十 KB 塞进上下文。上下文是有成本的。所以在 docstring 里我明确写了“默认 50 行”，引导模型养成加 LIMIT 的习惯。
资源：把不变的东西挂出去
工具是模型主动调用的，资源（resource）更像是一个可以被按需拉取的只读视图。适合放那些结构固定、不需要参数、会反复用到的内容。
@mcp.resource("schema://overview")
def schema_overview() -> str:
"""整个数据库的表结构与字段清单，客户端可随时读取。"""
with _connect() as conn:
tables = [
row["name"]
for row in conn.execute(
"select name from sqlite_master "
"where type='table' and name not like 'sqlite_%' order by name"
)
]
lines = []
for name in tables:
info = conn.execute(f'pragma table_info("{name}")').fetchall()
cols = ", ".join(f'{r["name"]}:{r["type"] or "any"}' for r in info)
lines.append(f"{name}({cols})")
return "n".join(lines)
URI 里用自定义的 schema:// 协议头，这是约定俗成的写法，表示这是本地资源而不是网络地址。客户端配置好之后，用户可以在界面里手动选择要不要把这个资源塞进上下文，主动权在他手上。
提示模板：把重复的提问固化下来
做数据分析的时候，我每次都要重复一遍那套检查流程，很烦。提示模板（prompt）就是干这个用的，把常用的提问套路写死。
@mcp.prompt()
def data_audit(table: str) -> str:
"""生成一份针对指定表的数据体检清单。"""
return (
f"请对 {table} 表做一次数据体检，按顺序完成下面几件事：n"
f"1. 执行 select count(*) from {table}，报告总行数。n"
f"2. 查看字段结构，逐个字段统计空值数量。n"
f"3. 找出可能的主键，检查是否存在重复值。n"
f"4. 对数值型字段给出最小值、最大值和平均值。n"
f"每完成一步都要说明你执行了什么 SQL、得到了什么结果。n"
f"最后按严重程度列出三条最需要处理的数据问题。"
)
模板返回的是一段纯文本，客户端会把它当成用户消息发出去。注意我在里面明确要求模型报告执行过的 SQL，这样你能验证它是真查了还是编的。
挂到客户端上
写完不代表能用，还要让客户端知道怎么启动这个进程。以 Claude Desktop 为例，配置文件的位置在 macOS 上是 ~/Library/Application Support/Claude/claude_desktop_config.json，Windows 上是 %APPDATA%Claudeclaude_desktop_config.json。
{
"mcpServers": {
"sqlite-explorer": {
"command": "uv",
"args": [
"--directory",
"/Users/you/projects/sqlite-mcp",
"run",
"server.py"
]
}
}
}
--directory 一定要写绝对路径，客户端启动子进程时的工作目录不是你以为的那个。我最早就是用了相对路径，客户端界面上工具列表一直是空的，日志里只有一行看不懂的启动失败。
改完配置必须完全退出客户端再重开，不是关窗口，是退出进程。这一点几乎每个新手都会踩一次。
Cursor 的配置格式类似，写在 .cursor/mcp.json 里，项目级和全局级都可以。
调试的时候看哪里
mcp dev server.py 是最常用的手段，上面已经演示过。除此之外还有两个地方值得看。
配置好客户端之后，MCP 的日志会写到 Claude Desktop 的 ~/Library/Logs/Claude/mcp-server-sqlite-explorer.log。工具没出现在列表里、调用卡住、返回异常，答案基本都在这个文件里。
另一个是直接在终端跑 stdio。把一段 JSON-RPC 消息喂进去，看输出：
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | uv run server.py
正常的话你会看到一串工具定义。如果什么都没有，或者输出了一段不是 JSON 的文本，那就说明你的代码往 stdout 里写了不该写的东西。
几个真会让人卡住的地方
print 是杀手
stdio 传输模式下，标准输出就是协议通道。你在代码里随手敲一句 print("查到了", len(rows))，这行字会直接插进 JSON-RPC 的消息流里，客户端解析失败，报一个跟 print 毫无关系的错。我第一次遇到的时候，客户端只说 Invalid JSON，我盯着 SQL 查了半天才发现是调试用的 print 忘了删。
想输出日志就老老实实用 logging，并且把 handler 指向 stderr：
import logging
import sys
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format="%(asctime)s %(levelname)s %(message)s",
)
stderr 是安全的，客户端会把它当日志收走，不会污染协议。
类型注解不能偷懒
FastMCP 靠类型注解生成 JSON Schema。如果你写 def run_query(sql)，生成出来的 schema 里这个参数就是空的，模型不知道要传什么，调用经常失败。如果写 sql: Any，schema 出来也是无类型的，模型只能靠猜。
老老实实写 str、int、float、bool。有默认值的参数会被标成可选，没默认值的就是必填，这个规则和 Python 函数本身一致。
别让工具名和参数名说方言
用英文命名。中文工具名在很多客户端的界面里会显示成乱码，参数名也是一样。描述文字用中文没问题，模型理解得很好，但标识符保持 ASCII 是更稳妥的选择。
返回体要有节制
上面提到的 limit 只是一部分。如果某张表有 200 个字段，一行数据就能占掉上千个 token。我一般还会在 run_query 里加一层粗略的字节数检查，超过阈值就直接截断并标注，宁可让模型多查几次，也不要把上下文撑爆。
打包出去给别人用
自己用的话，uv run server.py 就够了。想让别人一条命令跑起来，可以打包成包然后走 uvx。
uv build
在 pyproject.toml 里加一个脚本入口，指向 server 的启动函数：
[project.scripts]
sqlite-explorer = "sqlite_mcp.server:main"
之后别人只要在配置里写：
{
"mcpServers": {
"sqlite-explorer": {
"command": "uvx",
"args": ["sqlite-explorer", "--db", "/path/to/their.db"]
}
}
}
uvx 会自动拉包、建临时环境、执行入口，用的人不需要关心 Python 版本和依赖冲突。这种分发方式对非开发者的朋友特别友好，他要做的只是改一个路径。
最后说两句
整篇文章的代码加起来一百多行，涵盖了一个可用的 MCP Server 需要的全部要素：工具定义、参数校验、资源暴露、提示模板、错误处理、客户端挂载。真正花时间的地方不在写代码，而在想清楚每个工具该给模型什么信息、出错时该告诉它什么。
换个数据源，这套结构完全可以照搬。把 SQLite 换成 HTTP 接口、文件系统、内部运维系统，把只读查询换成具体的业务动作，剩下的部分——装饰器、类型注解、docstring、异常消息——写法是一模一样的。MCP 的价值就在这里：协议层把通信的脏活干完了，你只需要专心描述你的能力。
下一步最值得做的，是给 run_query 加一层查询结果缓存，把反复出现的相同 SQL 直接命中内存。在数据探索阶段模型会重复问同样的问题，这一层省下的时间相当可观。
