首页 › tutorials › 数据库自省MCP服务器构建实战

Claude Code MCP 服务器开发：构建一个会"自省"的数据库工具

概述

本教程聚焦 MCP（Model Context Protocol）服务器开发中最实用、却极少被系统讲解的一类——数据库自省型 MCP 服务器。我们将从零构建一个 PostgreSQL MCP 服务器，让 Claude Code 能像 DBA 一样读懂库结构、安全地只读查询、受控地写入数据，并把库表结构以 Resource 形式暴露给模型。读者将掌握 Tool / Resource / Prompt 三原语的合理分工、读写分离的安全护栏、stdio 与 Streamable HTTP 双传输的部署方式，以及用 MCP Inspector 全链路联调的方法。

适用人群：已有 Claude Code 基础、用过一两个现成 MCP 服务器，想亲手造一个面向真实数据源、能上生产环境工具的后端工程师与 AI 应用开发者。

前置条件

Claude Code 已安装（claude --version 可用），了解 /mcp 与 claude mcp 子命令的基本用法。
Python 3.11+ 与 uv（推荐）或 pip。本教程使用官方 mcp Python SDK，其中内置的 FastMCP 已由社区捐赠并入 Anthropic 官方 SDK。
一个可访问的 PostgreSQL 13+ 实例。本地可用 Docker 一行起一个： bash docker run -d --name pg -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=shop -p 5432:5432 postgres:16
安装核心依赖： bash pip install "mcp[cli]" psycopg[binary] httpx
准备示例数据（一张订单表）： bash psql "postgresql://postgres:secret@localhost/shop" -c \ "CREATE TABLE orders (id serial PRIMARY KEY, customer text, amount numeric, status text, created_at timestamptz default now()); INSERT INTO orders(customer, amount, status) VALUES ('Alice', 120, 'paid'),('Bob', 88, 'pending'),('Carol', 240, 'paid'),('Alice', 17, 'refunded');"

第一步：理解数据库 MCP 服务器的设计哲学

很多人写 MCP 工具的第一反应是"包一个 execute_query(sql) 给模型随便跑"。这在演示阶段很爽，在生产环境却是灾难——模型会写出没有 LIMIT 的全表扫描、DELETE FROM orders 这样的灭顶语句、以及读写混杂的事务。

一个工程化的数据库 MCP 服务器应遵循三条原则：

读用读连接、写用写连接。读路径默认强制 LIMIT 与语句超时，绝不放行 DML；写路径必须显式开启事务且可回滚。
结构信息优先于数据。先让模型"看见"库的结构（自省），再让它"动手"查数据。这就是 Resource 的职责。
三原语分工：Resource 暴露结构（db://schema/orders），Tool 执行动作（查询/写入），Prompt 封装高频分析套路（"审计某张表的慢查询"）。

常见问题：为什么不用纯 RAG 把数据库喂给模型？解决：RAG 只能回答"向量相似"的预设片段，无法应对"统计昨天未付款订单金额"这类即席聚合。MCP 给的是计算能力（让模型自己写 SQL 并执行），而非静态文本。

第二步：用 FastMCP 搭起服务器骨架

新建 db_mcp_server.py，先用三个只读工具把"自省 + 查询"打通：

# db_mcp_server.py
import os, re, json
import psycopg
from psycopg.rows import dict_row
from mcp.server.fastmcp import FastMCP

DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://postgres:secret@localhost/shop")
READ_ROW_LIMIT = 200

mcp = FastMCP("db-explorer")

async def _fetch(sql: str, params: tuple = ()):
    """统一只读执行入口，强制 200 行上限。"""
    async with await psycopg.AsyncConnection.connect(DATABASE_URL, row_factory=dict_row) as conn:
        async with conn.cursor() as cur:
            await cur.execute(sql, params)
            return await cur.fetchall()

@mcp.tool()
async def list_tables() -> str:
    """列出 public schema 下所有表名。"""
    rows = await _fetch(
        "SELECT table_name FROM information_schema.tables "
        "WHERE table_schema='public' ORDER BY table_name")
    return json.dumps([r["table_name"] for r in rows])

@mcp.tool()
async def describe_table(table_name: str) -> str:
    """返回某张表的列名、类型、是否可空、默认值。"""
    rows = await _fetch(
        """SELECT column_name, data_type, is_nullable, column_default
             FROM information_schema.columns
            WHERE table_schema='public' AND table_name=%s
            ORDER BY ordinal_position""", (table_name,))
    return json.dumps(rows, default=str)

_READONLY_RE = re.compile(r"^\s*(select|with|explain)\b", re.I)
_WRITE_RE     = re.compile(r"^\s*(insert|update|delete|merge|drop|alter|truncate|grant)\b", re.I)

@mcp.tool()
async def run_readonly_query(sql: str) -> str:
    """执行只读 SELECT/WITH/EXPLAIN，最多返回 200 行。任何写操作都会被拦截。"""
    if not _READONLY_RE.match(sql) or _WRITE_RE.match(sql):
        return "BLOCKED: 仅允许 SELECT/WITH/EXPLAIN。"
    wrapped = f"SELECT * FROM ({sql}) AS _q LIMIT {READ_ROW_LIMIT}"
    rows = await _fetch(wrapped)
    return json.dumps(rows, default=str)

if __name__ == "__main__":
    mcp.run()  # 默认 stdio 传输

预期结果

直接运行 python db_mcp_server.py 会看到进程"挂起"等待 stdin——这是正常的，stdio MCP 服务器就是靠标准输入输出收发 JSON-RPC。下一步会用 Inspector 和 Claude Code 来驱动它。

常见问题：启动报 KeyError: 'DATABASE_URL'。解决：把环境变量传进去，或修改默认值。后续用 claude mcp add 时务必用 env 前缀注入。

第三步：用 MCP Inspector 全链路联调

在接入 Claude Code 之前，先用官方调试器把每个工具跑通，这是省时间的黄金法则：

npx @modelcontextprotocol/inspector python db_mcp_server.py

浏览器会打开一个本地面板：

在左侧 Tools 列表里能看到 list_tables、describe_table、run_readonly_query。
点击 describe_table，填 table_name=orders，点 Run Tool，右侧应返回 JSON 列信息。
用 run_readonly_query 跑 SELECT customer, sum(amount) FROM orders GROUP BY customer，验证聚合正常。
故意提交 DELETE FROM orders，应立刻收到 BLOCKED: 仅允许... 的返回——证明护栏生效。

flowchart LR
    A[Inspector UI] -->|JSON-RPC over stdio| B[db_mcp_server.py]
    B --> C{语句分类}
    C -->|SELECT/WITH| D[只读连接\n强制 LIMIT]
    C -->|INSERT/UPDATE/DELETE| E[BLOCKED\n拦截写操作]
    C -->|DDL/DROP| E
    D --> F[(PostgreSQL shop)]
    F -->|结果集| B --> A

常见问题：Inspector 里工具能跑，但 Claude Code 里报 server exited。解决：八成是环境变量没带上。Inspector 在你当前 shell 启动子进程能继承环境；而 Claude Code 注册的命令是独立进程，必须显式 env。

第四步：补齐 Resource 与 Prompt，让服务器"立体"

只读查询只是"手"，结构暴露（Resource）和复用套路（Prompt）是"眼"和"脑"。把它们加上：

@mcp.resource("db://schema/{table}")
async def table_schema(table: str) -> str:
    """把表结构作为资源暴露，模型可在上下文中直接引用 db://schema/orders。"""
    rows = await _fetch(
        "SELECT column_name, data_type FROM information_schema.columns "
        "WHERE table_name=%s ORDER BY ordinal_position", (table,))
    return json.dumps(rows, default=str)

@mcp.prompt()
def analyze_table(table_name: str) -> str:
    """一键生成"分析某张表"的提示词，引导模型按固定套路工作。"""
    return (
        f"你是资深 DBA。请分析表 `{table_name}`：\n"
        f"1) 先用 describe_table 读取结构；\n"
        f"2) 用 run_readonly_query 抽样 10 行；\n"
        f"3) 指出可能的索引缺失与数据质量风险；\n"
        f"4) 给出 3 条优化建议。"
    )

分工的意义：当 Claude Code 在对话里需要"orders 表长什么样"时，它可以读 db://schema/orders 这个 Resource（一次性、结构化），而不是反复调用 describe_table 工具（每次都要一次往返）。Prompt 则把"如何审计一张表"的最佳实践固化下来，团队成员用 /analyze-table orders 一键复用，避免每个人各写各的提示词。

第五步：受控写入——带护栏的事务工具

只读服务器对企业足够安全，但很多时候模型需要真正改数据（如打标签、补字段）。关键是在"能力"和"护栏"之间设防火墙：

@mcp.tool()
async def run_write(sql: str) -> str:
    """执行单条 INSERT/UPDATE/DELETE，自动包在事务里，失败即回滚。禁止 DDL。"""
    if _READONLY_RE.match(sql) or not re.match(r"^\s*(insert|update|delete|merge)\b", sql, re.I):
        return "BLOCKED: 仅允许单条 INSERT/UPDATE/DELETE/MERGE。"
    async with await psycopg.AsyncConnection.connect(DATABASE_URL) as conn:
        async with conn.transaction():
            async with conn.cursor() as cur:
                await cur.execute(sql)
                return f"OK, affected rows: {cur.rowcount}"

设计要点： - 事务包裹：async with conn.transaction() 保证要么全成功、要么全回滚。 - 白名单动词：只放行 DML，彻底封死 DROP/ALTER/TRUNCATE。 - 单语句：拒绝分号多语句，杜绝注入叠加。

更严格的做法是结合 Elicitation（MCP 的"反向询问"机制）：写工具被调用时弹回一个确认请求给用户，确认后才真正 commit。这把"危险操作必须人确认"落到了协议层，而非依赖模型自觉。

sequenceDiagram
    participant CC as Claude Code
    participant S as MCP 服务器
    participant U as 用户
    participant DB as PostgreSQL
    CC->>S: run_write("UPDATE orders SET status='paid' WHERE id=1")
    S->>S: 动词白名单校验
    S->>U: Elicitation: 确认执行此 UPDATE?
    U->>S: 确认
    S->>DB: BEGIN; UPDATE ...; COMMIT
    DB-->>S: affected=1
    S-->>CC: OK, affected rows: 1

常见问题：模型仍偶尔写出危险语句。解决：再加一道库侧护栏——给 MCP 用一个 default_transaction_read_only=on 的低权限角色做读，写角色只授予目标表的最小 DML 权限。协议层 + 库权限，双保险。

第六步：接入 Claude Code（stdio 项目级）

让 Claude Code 认识这台服务器。推荐用项目级 .mcp.json，便于团队共享、纳入版本管理：

{
  "mcpServers": {
    "db-explorer": {
      "command": "python",
      "args": ["/绝对路径/db_mcp_server.py"],
      "env": { "DATABASE_URL": "postgresql://postgres:secret@localhost/shop" }
    }
  }
}

放到仓库根目录后，Claude Code 首次进入会询问是否信任该 MCP 服务器，确认即可。也可以用 CLI 一键注册（local 作用域）：

claude mcp add db-explorer -- env DATABASE_URL=postgresql://postgres:secret@localhost/shop python ./db_mcp_server.py

验证：

claude mcp list          # 应出现 db-explorer ✓ Connected
claude mcp get db-explorer

进入会话后试一句自然语言：

帮我看看 orders 表里哪个客户消费最多，列出前三名。

Claude Code 会自动调用 describe_table → run_readonly_query，返回结果并解释。整个过程模型自主选工具，你无需手写 SQL。

第七步：升级为远程服务器（Streamable HTTP + OAuth）

当团队多人、或要在云端（如一个内部数据 API）共享同一台 MCP 服务器时，stdio 就不够了——每次都要在客户端拉起子进程。2025-03-26 规范起，Streamable HTTP 取代了旧的 HTTP+SSE，成为远程 MCP 的标准传输；2026-07-28 的规范候选版进一步强化了有状态会话、水平扩展与企业级 OAuth 2.1。

把服务器切成 HTTP 模式只需改最后一行：

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

在 Claude Code 中以远程方式接入：

claude mcp add --transport http db-remote https://db.example.com/mcp

若服务器启用了 OAuth 2.1（强烈建议公网部署时启用），Claude Code 会在首次连接时弹出浏览器完成授权码流程，拿到 token 后写入本地凭证。这条路径让"公司内部 BI 数据库"也能安全地变成团队所有 Claude Code 用户的共享上下文，而无需任何人本地配置连接串。

flowchart TB
    subgraph 客户端
      CC1[Claude Code 工程师A]
      CC2[Claude Code 工程师B]
    end
    CC1 -->|HTTPS + Bearer Token| GW[MCP Gateway\nOAuth 2.1]
    CC2 -->|HTTPS + Bearer Token| GW
    GW --> RLB[远程 db-explorer\nStreamable HTTP]
    RLB --> PG[(PostgreSQL 生产库\n只读副本 + 写副本)]
    GW -.-> IDP[企业 IdP / SSO]

常见问题：远程模式连接一直 401。解决：确认服务器声明了 OAuth 2.1 的 metadata endpoint（/.well-known/oauth-authorization-server），且 redirect URI 在客户端允许列表里。Claude Code 严格遵循 RFC 9700（OAuth 2.1 草案），任何不规范的 issuer 都会被拒。

实战案例：构建一个"客服自助分析"数据库 MCP 服务器

把前面六步串成一个可交付项目。目标：让一线客服用自然语言问"昨天有多少退款单、金额多少"，由 Claude Code + 数据库 MCP 直接作答，且零写权限、零全表扫描。

目录结构

db-mcp/
├── .mcp.json
├── pyproject.toml
├── db_mcp_server.py
├── tests/
│   └── test_guardrails.py
└── README.md

核心服务器（综合前文代码，此处给出完整可运行版的关键加固点）：

# db_mcp_server.py —— 生产加固版片段
import os, re, json, psycopg
from psycopg.rows import dict_row
from mcp.server.fastmcp import FastMCP

READ_DB  = os.environ["READ_DB_URL"]   # 只读副本
WRITE_DB = os.environ.get("WRITE_DB_URL", READ_DB)
ROW_CAP  = 200
ALLOWED  = {"orders", "customers", "refunds"}  # 表白名单

mcp = FastMCP("cs-analytics")

@mcp.tool()
async def run_readonly_query(sql: str) -> str:
    if not re.match(r"^\s*(select|with|explain)\b", sql, re.I):
        return "BLOCKED"
    # EXPLAIN 先算成本，成本过高直接拒
    async with await psycopg.AsyncConnection.connect(READ_DB, row_factory=dict_row) as c:
        async with c.cursor() as cur:
            await cur.execute(f"EXPLAIN {sql}")
            cost = max((float(r.get("plan rows", 0)) for r in await cur.fetchall()), default=0)
            if cost > 100_000:
                return f"BLOCKED: 预估扫描 {cost} 行，请加 WHERE 或加索引。"
            await cur.execute(f"SELECT * FROM ({sql}) AS _q LIMIT {ROW_CAP}")
            return json.dumps(await cur.fetchall(), default=str)

护栏单测（保证 BLOCKED 逻辑不回归）：

# tests/test_guardrails.py
import asyncio, db_mcp_server as m

async def main():
    assert "BLOCKED" in await m.run_readonly_query("DELETE FROM orders")
    assert "BLOCKED" in await m.run_readonly_query("SELECT * FROM orders; DROP TABLE orders")
    out = await m.run_readonly_query("SELECT count(*) FROM orders")
    assert out.strip().startswith("[")  # 正常返回 JSON

asyncio.run(main())
print("guardrails OK")

运行 python tests/test_guardrails.py，预期输出 guardrails OK。

部署后效果：客服在 Claude Code 里问"退款最多的三位客户"，模型依次调用 describe_table（看结构）→ run_readonly_query（带 LIMIT 的聚合）→ 自然语言总结。EXPLAIN 成本护栏会在有人写出 SELECT * FROM huge_table 时提前拦截，只读副本保证查询不影响主库交易，表白名单把越权查其他库的可能直接掐断。

训练营适用性评估

适用类型：中后端工程师进阶营、AI 应用开发营、企业数据智能营。不适合纯前端或零编程基础营（涉及异步、SQL、协议三层知识）。
教学时长：建议 半天到一天（4–6 小时）。前 3 小时打通 stdio + 三原语 + Inspector；后 1–3 小时做实战案例（含护栏单测与远程 OAuth 部署）。
难度曲线：第 1–4 步平缓（照搬代码即可运行）；第 5 步（写入护栏）是难点，建议讲师现场演示一次 Elicitation 确认流程；第 7 步（远程 + OAuth）可作为加分项，仅对有运维背景的学员展开。
交付物：每位学员产出一台可接入自己 Claude Code 的数据库 MCP 服务器 + .mcp.json + 护栏单测，可直接用于本职工作。

参考来源

Building MCP Servers for PostgreSQL: Schema Introspection and Safe Operations — 数据库 MCP 服务器的 schema 自省与安全读写实践，本教程护栏设计的直接参考。
Complete Guide to MCP (Model Context Protocol) in 2026 — Architecture & Implementation — 2026 年 MCP 全景：Streamable HTTP 传输、OAuth 2.1 认证与生产部署架构。
The 2026-07-28 MCP Specification Release Candidate — 官方最新规范候选版，有状态会话、水平扩展与 Agent 通信能力说明。
FastMCP · Resources & Templates — Resource 与资源模板的官方用法，对应本教程第四步结构暴露设计。
Query Databases From AI via MCP (RapidDev Tutorial) — 自然语言查询 PostgreSQL/SQLite 的端到端示例，验证 Tool/Resource 分工思路。

[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入