<!-- title: 数据库自省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 服务器应遵循三条原则：

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

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

---

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

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

```python
# 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 之前，先用官方调试器把每个工具跑通，这是省时间的黄金法则：

```bash
npx @modelcontextprotocol/inspector python db_mcp_server.py
```

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

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

```mermaid
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）是"眼"和"脑"。把它们加上：

```python
@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` 一键复用，避免每个人各写各的提示词。

---

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

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

```python
@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。这把"危险操作必须人确认"落到了协议层，而非依赖模型自觉。

```text
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`**，便于团队共享、纳入版本管理：

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

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

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

验证：

```bash
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 模式只需改最后一行：

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

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

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

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

```mermaid
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
```

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

```python
# 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 逻辑不回归）：

```python
# 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` 时提前拦截，只读副本保证查询不影响主库交易，表白名单把越权查其他库的可能直接掐断。

---

## 延伸阅读

- **Model Context Protocol 官方规范与 Python SDK**：`modelcontextprotocol.io` 与 GitHub `modelcontextprotocol/python-sdk`，权威的三原语定义与 FastMCP 最新 API。
- **MCP Inspector 调试指南**：`modelcontextprotocol.io/docs/tools/inspector`，掌握断点、回放、协议日志查看。
- **FastMCP 完整文档**：`gofastmcp.com`，涵盖 Context 进度上报、资源订阅（subscriptions）、任务（tasks）等高级特性。
- **2026 MCP Roadmap 与规范候选版**：`blog.modelcontextprotocol.io`，跟踪 Streamable HTTP、OAuth 2.1、会话可扩展性与 Agent 通信的演进方向。
- **Claude Code MCP 配置参考**：`docs.anthropic.com` 下 "MCP" 章节，`.mcp.json` 三作用域（local/project/user）与 `claude mcp` 子命令全表。

---

## 训练营适用性评估

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

---

## 参考来源

1. [Building MCP Servers for PostgreSQL: Schema Introspection and Safe Operations](https://www.arika.dev/blog/mcp/mcp-postgresql-protocol/) — 数据库 MCP 服务器的 schema 自省与安全读写实践，本教程护栏设计的直接参考。
2. [Complete Guide to MCP (Model Context Protocol) in 2026 — Architecture & Implementation](https://dev.to/x4nent/complete-guide-to-mcp-model-context-protocol-in-2026-architecture-implementation-and-4a11) — 2026 年 MCP 全景：Streamable HTTP 传输、OAuth 2.1 认证与生产部署架构。
3. [The 2026-07-28 MCP Specification Release Candidate](https://blog.modelcontextprotocol.io/posts/2026-07-28-release-candidate/) — 官方最新规范候选版，有状态会话、水平扩展与 Agent 通信能力说明。
4. [FastMCP · Resources & Templates](https://gofastmcp.com/servers/resources) — Resource 与资源模板的官方用法，对应本教程第四步结构暴露设计。
5. [Query Databases From AI via MCP (RapidDev Tutorial)](https://www.rapidevelopers.com/mcp-tutorial/how-to-use-mcp-to-query-databases-from-ai) — 自然语言查询 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学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)