sqlite-mcp-server

AI-student2024/sqlite-mcp-server

3.1

If you are the rightful owner of sqlite-mcp-server and would like to certify it and/or have it hosted online, please leave a comment on the right or send an email to henry@mcphub.com.

A minimal MCP-based SQLite read-only service designed for direct connection with Cursor/Cline agents.

Tools
5
Resources
0
Prompts
0

sqlite-mcp-server

一个基于 MCP 的 SQLite 只读服务，满足"模式1：Cursor/Cline 直连 MCP Server"的最小闭环。

项目概述

目标

  • 快速验证：建立可落地的 MCP Server 方案
  • 安全可靠：提供只读数据库访问能力
  • 多 Agent 支持：供 Cursor/Cline 等 Agent 即插即用
  • 学习实践：满足学习与多 Agent 落地双重需求

实现思路

  • 薄服务、厚智能：MCP Server 专注稳定/安全的数据库能力
  • 标准接口：遵循 MCP 协议规范，提供标准工具接口
  • 安全优先：只读策略、SQL 校验、审计日志
  • 性能优化：分页、超时、连接管理

技术架构

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Cursor/Cline  │───▶│  sqlite-mcp-   │───▶│   SQLite DB     │
│   (Agent)       │    │   server        │    │   (Sakila)      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                              │
                              ▼
                       ┌─────────────────┐
                       │   Audit Log     │
                       │   (JSON)        │
                       └─────────────────┘

运行

  • 确保已安装依赖（mcp[cli], httpx 等），并在虚拟环境内：
  • 启动（stdio 模式）：
python sqlite-mcp-server.py

或在 mcp_servers.json 中配置为可发现的本地 server。

工具（Tools）

  • connect_database
    • 入: { db_url, role:"readonly" }
    • 出: { connection_id, db_engine, tables:[{name,row_count}], ttl_s }
  • get_schema_summary
    • 入: { connection_id, include_views?:bool }
    • 出: { tables:[{name, columns:[{name,type,nullable,pk,fk,ref_table,ref_column}], indexes:[...]}], updated_at }
  • get_ddl
    • 入: { connection_id, tables?:[string] }
    • 出: { ddls:[{table, create_sql}] }
  • explain_sql
    • 入: { connection_id, sql }
    • 出: { ok, readonly_ok, plan?, reason? }
  • execute_sql
    • 入: { connection_id, sql, page?:number, page_size?:number, timeout_ms?:number }
    • 出: { columns, rows, row_count, page, page_size, execution_ms }

工作流程

MCP Server 遵循以下完整的工作流程，确保安全、高效的数据库访问：

1. 数据库连接阶段 (connect_database)

输入：数据库 URL 地址

{
  "db_url": "D:/AgentsDEV/mcp-test/rag-sqlite/90-文档-Data/sakila/sakila.db",
  "role": "readonly"
}

处理过程：

  • 以只读模式打开 SQLite 连接
  • 生成唯一的 connection_id
  • 获取数据库表列表和行数统计
  • 记录连接审计日志

输出：连接信息和表概览

{
  "connection_id": "bed29ae3-0456-4fd8-a2d6-0abff9f3fec3",
  "db_engine": "sqlite",
  "tables": [
    {"name": "film", "row_count": 1000},
    {"name": "actor", "row_count": 200}
  ],
  "ttl_s": 3600
}

2. 表结构获取阶段 (get_schema_summary / get_ddl)

Schema 摘要：

  • 获取所有表的详细结构（列名、类型、约束）
  • 获取索引信息
  • 可选择包含视图

DDL 生成：

  • 生成指定表的 CREATE 语句
  • 支持单表或多表 DDL 获取

3. 查询验证阶段 (explain_sql)

安全性验证：

# 检查是否为只读查询
_SQL_FORBIDDEN_PATTERNS = re.compile(
    r"\b(INSERT|UPDATE|DELETE|REPLACE|CREATE|ALTER|DROP|TRUNCATE|ATTACH|DETACH|VACUUM|PRAGMA|REINDEX|ANALYZE|BEGIN|END|COMMIT|ROLLBACK|SAVEPOINT|RELEASE)\b",
    re.IGNORECASE,
)

验证内容：

  • ✅ 仅允许 SELECT 或 EXPLAIN SELECT
  • ✅ 拒绝所有 DML/DDL 操作
  • ✅ 防多语句攻击
  • ✅ 语法正确性检查

执行计划分析：

  • 使用 EXPLAIN QUERY PLAN 分析查询性能
  • 预估执行时间和资源消耗

4. 查询执行阶段 (execute_sql)

性能控制：

  • 超时控制：默认 5 秒超时
  • 分页处理：自动添加 LIMIT/OFFSET
  • 结果格式化：返回列名和行数据

输出示例：

{
  "columns": ["title", "rating", "rental_rate"],
  "rows": [
    ["ACADEMY DINOSAUR", "PG", 0.99],
    ["ACE GOLDFINGER", "G", 4.99]
  ],
  "row_count": 1000,
  "page": 1,
  "page_size": 5,
  "execution_ms": 19
}

5. 审计日志记录

每个操作都会记录到 server_audit.log：

{
  "event": "execute_sql",
  "ts": "2025-08-08T15:12:10Z",
  "connection_id": "bed29ae3-0456-4fd8-a2d6-0abff9f3fec3",
  "sql": "SELECT f.title, COUNT(*) AS rentals...",
  "execution_ms": 19,
  "ok": true
}

自然语言转 SQL 流程

在这个项目中，自然语言转 SQL 的过程由 Agent（如 Cursor）处理，MCP Server 提供标准化的数据库工具支持。以下是完整的转换流程：

转换流程图

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   用户输入      │───▶│   Agent         │───▶│  MCP Server     │───▶│   SQLite DB     │
│   自然语言      │    │   (Cursor)      │    │   (工具提供)     │    │   (数据存储)     │
└─────────────────┘    └─────────────────┘    └─────────────────┘    └─────────────────┘
                              │                        │                        │
                              ▼                        ▼                        ▼
                       ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
                       │   1. 理解意图   │    │   2. 获取 Schema │    │   3. 验证 SQL   │
                       │   2. 生成 SQL   │    │   3. 执行查询   │    │   4. 返回结果   │
                       │   3. 解释结果   │    │   4. 审计日志   │    │                 │
                       └─────────────────┘    └─────────────────┘    └─────────────────┘

详细转换步骤

步骤 1：用户输入自然语言
用户：请查询所有电影标题
步骤 2：Agent 理解并转换

Agent 处理过程：

  1. 理解用户意图：用户想要获取电影标题列表
  2. 分析数据库结构：通过 MCP 工具获取 schema
  3. 生成 SQL：基于 schema 信息生成合适的 SQL
步骤 3：获取数据库结构
// Agent 调用
get_schema_summary({
  "connection_id": "xxx",
  "include_views": false
})

// 返回结果
{
  "tables": [
    {
      "name": "film",
      "columns": [
        {"name": "film_id", "type": "INTEGER", "pk": true},
        {"name": "title", "type": "TEXT", "nullable": false},
        {"name": "rating", "type": "TEXT"},
        {"name": "rental_rate", "type": "REAL"}
      ]
    }
  ]
}
步骤 4：生成 SQL
SELECT title FROM film
步骤 5：验证 SQL
// Agent 调用
explain_sql({
  "connection_id": "xxx",
  "sql": "SELECT title FROM film"
})

// 返回验证结果
{
  "ok": true,
  "readonly_ok": true,
  "plan": "SCAN TABLE film"
}
步骤 6：执行查询
// Agent 调用
execute_sql({
  "connection_id": "xxx",
  "sql": "SELECT title FROM film",
  "page": 1,
  "page_size": 1000
})

// 返回结果
{
  "columns": ["title"],
  "rows": [
    ["ACADEMY DINOSAUR"],
    ["ACE GOLDFINGER"],
    ["ADAPTATION HOLES"]
  ],
  "row_count": 1000
}

复杂查询示例

用户输入
用户：查询最受欢迎的电影（按租赁次数排序）
Agent 处理过程

1. 获取 Schema：

get_schema_summary({"connection_id": "xxx"})

2. 分析表关系：

  • film 表：电影信息
  • inventory 表：库存信息（film_id 关联）
  • rental 表：租赁记录（inventory_id 关联）

3. 生成 SQL：

SELECT f.title, COUNT(*) AS rental_count
FROM rental r
JOIN inventory i ON r.inventory_id = i.inventory_id
JOIN film f ON i.film_id = f.film_id
GROUP BY f.film_id, f.title
ORDER BY rental_count DESC
LIMIT 10

4. 验证并执行：

explain_sql({"connection_id": "xxx", "sql": "..."})
execute_sql({"connection_id": "xxx", "sql": "..."})

职责分工

组件职责具体功能
Agent (Cursor)自然语言理解、SQL 生成、结果解释• 理解用户意图
• 分析数据库结构
• 生成 SQL 语句
• 解释查询结果
MCP Server数据库连接、SQL 验证、查询执行• 提供数据库工具接口
• 验证 SQL 安全性
• 执行查询并返回结果
• 记录审计日志

设计优势

1. 安全性
  • MCP Server 严格验证 SQL，防止危险操作
  • 只读策略确保数据安全
2. 灵活性
  • Agent 可以根据不同场景生成不同的 SQL
  • 支持复杂的多表关联查询
3. 可扩展性
  • 可以支持多种 Agent 和不同的自然语言处理模型
  • 遵循 MCP 协议，工具接口标准化
4. 可观测性
  • 完整的审计日志记录
  • 详细的性能指标监控

实际使用示例

在 Cursor 中，你可以直接使用自然语言：

"请连接到 Sakila 数据库，查询所有电影标题"

"连接数据库后，请统计每个国家的顾客数量，按数量降序排列"

"查询最长的电影是哪一部"

Agent 会自动：

  1. 连接数据库
  2. 获取表结构
  3. 生成合适的 SQL
  4. 验证并执行
  5. 返回结果并解释

这种"薄服务、厚智能"的设计让系统更加模块化和安全！🎯

策略

  • 仅允许 SELECT/WITH...SELECT/EXPLAIN SELECT，拒绝 INSERT/UPDATE/DELETE/CREATE/ALTER/DROP/... 等。
  • 默认分页：page_size=1000。超时：timeout_ms=5000（SQLite progress_handler 实现）。
  • 审计：server_audit.log 记录请求元数据与耗时。

示例（Sakila）

# 1) 连接（Windows 示例）
connect_database {"db_url":"D:/AgentsDEV/mcp-test/rag-sqlite/90-文档-Data/sakila/sakila.db","role":"readonly"}

# 2) 结构摘要
get_schema_summary {"connection_id":"..."}

# 3) 取 DDL
get_ddl {"connection_id":"...","tables":["film","actor"]}

# 4) Explain 校验
explain_sql {"connection_id":"...","sql":"SELECT * FROM film LIMIT 5"}

# 5) 分页执行
execute_sql {"connection_id":"...","sql":"SELECT * FROM film","page":1,"page_size":5}

Cursor 配置

在你的 mcp.json 中配置：

{
  "sqlite-mcp-server": {
    "command": "uv",
    "args": [
      "--directory",
      "D:/AgentsDEV/mcp-test/sqlite-mcp-server",
      "run",
      "sqlite-mcp-server.py"
    ]
  }
}

功能特点

  • ✅ 只读安全：仅允许 SELECT 查询，防止数据修改
  • ✅ 分页支持：大结果集自动分页，避免内存溢出
  • ✅ 超时控制：防止长时间查询阻塞
  • ✅ 审计日志：记录所有操作，便于监控和调试
  • ✅ Windows 兼容：自动处理路径格式问题
  • ✅ 错误处理：详细的错误信息，便于问题排查

使用场景

  1. 数据查询：执行只读 SQL 查询
  2. 结构分析：获取数据库表结构和 DDL
  3. 性能优化：通过 EXPLAIN 分析查询计划
  4. 安全审计：记录所有数据库操作

开发推进方案

当前实现符合 模式1 要求：

  • ✅ MCP Server 基础能力
  • ✅ 只读策略与安全校验
  • ✅ Cursor/Cline 联通
  • ✅ 审计与限流
  • ✅ 基础安全策略

为 模式2 预留扩展空间：

  • RAG 向量库与 NL2SQL
  • 可视化与报告生成
  • 多 Agent 支持

项目结构

sqlite-mcp-server/
├── .venv/                    # Python 虚拟环境
├── .git/                     # Git 版本控制
├── __pycache__/              # Python 缓存文件
├── resources/                 # 资源文件目录（历史遗留）
│   ├── ddl_full_*.json       # DDL 资源文件
│   └── query_*.json          # 查询结果资源文件
├── .gitignore                # Git 忽略文件配置
├── .python-version           # Python 版本声明
├── main.py                   # 入口文件（备用）
├── pyproject.toml            # 项目配置和依赖
├── README.md                 # 项目文档
├── server_audit.log          # 审计日志文件
├── sqlite-mcp-server.py      # 主程序文件
└── uv.lock                   # 依赖锁定文件

核心文件说明

  • sqlite-mcp-server.py：主程序文件，包含所有 MCP 工具实现
  • pyproject.toml：项目配置，定义依赖（mcp[cli], httpx）
  • server_audit.log：审计日志，记录所有操作
  • resources/：历史遗留的资源文件目录
  • .venv/：Python 虚拟环境，包含所有依赖

依赖说明

  • mcp[cli]>=1.12.4：MCP 协议实现
  • httpx>=0.28.1：HTTP 客户端库
  • Python>=3.10：最低 Python 版本要求

开发环境

  • 包管理：使用 uv 进行依赖管理
  • 虚拟环境：.venv 目录
  • 版本控制：Git 仓库
  • 审计日志：JSON 格式的详细日志