编码 agent 读代码很在行,但让它直接连生产数据库,风险不在 SQL 写得好不好,而在它可能看到不该看的表、或者一条 UPDATE 就把数据改了。这个教程用一个只读 MCP server 把范围锁死:agent 只能列出表、查看表结构、执行 SELECT,单次最多返回 200 行。代码不到 60 行,Windows、macOS、Linux 都能跑。

准备环境

需要 Python 3.10 以上。建议用 uv 建一个独立环境,避免污染系统 Python;下面命令在项目目录执行。

uv venv
# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate
uv pip install mcp

造一个测试库

先用 SQLite 建一个小库,包含两张表和几行数据。后面所有示例都连这个文件,确认能跑通再换成你自己的库。

import sqlite3
con = sqlite3.connect("demo.db")
con.executescript("""
CREATE TABLE IF NOT EXISTS customer(id INTEGER PRIMARY KEY, name TEXT, city TEXT);
CREATE TABLE IF NOT EXISTS orders(id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL, created_at TEXT);
INSERT INTO customer(name, city) VALUES ('张伟','上海'),('李娜','杭州'),('王强','深圳');
INSERT INTO orders(customer_id, amount, created_at) VALUES (1, 199.0, '2026-10-01'),(1, 88.5, '2026-10-12'),(2, 320.0, '2026-10-20');
""")
con.commit(); con.close()

写 server:只读连接 + 三个工具

关键点有三个:连接串用 mode=ro 打开只读文件;执行前检查语句是否以 SELECT 或 WITH 开头,并用 authorizer 再拦一层写操作;结果限制行数,避免一次把大表拉进上下文。FastMCP 会把带 @mcp.tool() 的函数自动暴露成工具。

import sqlite3
from pathlib import Path
from mcp.server.fastmcp import FastMCP

DB = Path("./demo.db").resolve()
mcp = FastMCP("sqlite-ro")

def connect():
    con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    con.row_factory = sqlite3.Row
    return con

@mcp.tool()
def list_tables() -> str:
    """列出数据库中的所有表。"""
    with connect() as con:
        rows = con.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").fetchall()
    return "\n".join(r["name"] for r in rows) or "(没有表)"

@mcp.tool()
def describe_table(table: str) -> str:
    """查看某张表的字段与类型。"""
    if not table.replace("_", "").isalnum():
        return "表名不合法"
    with connect() as con:
        rows = con.execute(f"PRAGMA table_info({table})").fetchall()
    return "\n".join(f"{r['name']} {r['type']}" for r in rows) or "表不存在"

@mcp.tool()
def query(sql: str, limit: int = 100) -> str:
    """执行一条只读 SQL(SELECT / WITH),最多返回 200 行。"""
    s = sql.strip().rstrip(";")
    if not (s.lower().startswith("select") or s.lower().startswith("with")):
        return "只允许 SELECT / WITH 查询"
    limit = max(1, min(int(limit), 200))
    con = connect()
    def deny(action, arg1, arg2, dbname, trigger):
        return sqlite3.SQLITE_DENY if action in (sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE, sqlite3.SQLITE_DROP_TABLE) else sqlite3.SQLITE_OK
    con.set_authorizer(deny)
    try:
        rows = con.execute(s).fetchmany(limit)
        if not rows:
            return "(无结果)"
        cols = rows[0].keys()
        lines = [" | ".join(str(c) for c in cols)]
        lines += [" | ".join("" if r[c] is None else str(r[c]) for c in cols) for r in rows]
        return "\n".join(lines)
    except Exception as e:
        return f"查询失败:{e}"
    finally:
        con.close()

if __name__ == "__main__":
    mcp.run()

先在终端里验证

MCP server 默认走 stdio,直接用 python 运行看起来会「卡住」,这是正常的:它在等 JSON-RPC 消息。更快的验证方式是用官方 Inspector,或者先单独跑一遍函数逻辑。最省事的做法是把 query() 里的核心代码复制到普通脚本里,确认能返回预期结果,再接 agent。

# 单独验证(不启动 MCP)
# python -c "from server import query; print(query('SELECT * FROM customer'))"

接进 Claude Code

Claude Code 用一条命令注册 stdio server,路径写绝对路径,避免 agent 在不同工作目录下找不到文件。注册后用 /mcp 查看连接状态,再让它「列出表并统计每个城市的客户数」做一次端到端验证。

claude mcp add sqlite-ro -- python D:/work/mcp-sqlite/server.py
# 进入会话后
/mcp

接进 Codex

Codex 在 ~/.codex/config.toml 里注册 MCP server。command 用虚拟环境里的 python,args 指向 server 的绝对路径;如果只想在某个项目里用,把配置写进该项目的 .codex/config.toml(以你当前版本支持的路径为准)。

[mcp_servers.sqlite-ro]
command = "D:/work/mcp-sqlite/.venv/Scripts/python.exe"
args = ["D:/work/mcp-sqlite/server.py"]

常见问题与边界

  • 报「No module named mcp」:agent 用的 python 不是你激活的那个虚拟环境,把 command 改成 .venv 里的绝对路径。
  • 查询一直超时:先确认 DB 路径是绝对路径,再检查文件是否被其他进程锁住;SQLite 只读连接不会阻塞读,但写事务会。
  • 返回内容太长:把 limit 调小,或者让 agent 先聚合再取明细;不要靠加大上下文硬塞。
  • 只读不等于安全:这个 server 仍然能读到库里所有表。生产库要先做视图或脱敏副本,别把整库直接暴露给 agent。
  • 不要把 server 暴露成公网 HTTP:stdio 由本机 agent 拉起,没有鉴权;一旦变成网络端口,就需要额外的认证与限流。
给 agent 开数据库,先决定它「不能做什么」,再决定它「能查什么」;顺序反了,权限就会越补越大。