shop-mcp

一个本地、只读的 MCP（模型上下文协议）服务器，让 AI 代理能够通过 stdio 传输分析 SQLite 电商数据库（shop.db）——包括客户、产品、订单和订单项。无需 HTTP 服务器，也无需单独的数据库进程：服务器直接打开 shop.db，并暴露两个小而通用的工具，供代理探索模式并运行自己的分析性 SQL。

使用官方 Python MCP SDK（PyPI 上的 mcp）构建。

项目结构

mcp-sql/
├── server.py                    # the MCP server (stdio transport)
├── shop.db                      # SQLite database (not modified by this project)
├── requirements.txt
├── .env.example
├── mcp-config.example.json
├── tests/
│   ├── conftest.py
│   ├── test_server.py           # unit tests (call tool functions directly)
│   └── test_stdio_integration.py# protocol-level test (spawns server.py over stdio)
└── README.md

数据库模式（实际在 shop.db 中找到的）

customers(id PK, first_name, last_name, email UNIQUE, phone, created_at)
products(id PK, name, category, price, stock_quantity, created_at)
orders(id PK, customer_id -> customers.id, order_date, status, total_amount)
order_items(id PK, order_id -> orders.id, product_id -> products.id, quantity, unit_price)

orders.status 被约束为：new、processing、shipped、completed、cancelled。products.category 目前有 5 个不同的值。外键：orders.customer_id → customers.id、order_items.order_id → orders.id、order_items.product_id → products.id。服务器在查询时从实时数据库（通过 sqlite_master / PRAGMA table_info / PRAGMA foreign_key_list）推导出所有这些信息——这里没有任何硬编码，因此如果 shop.db 被替换为另一个具有不同模式的文件，get_database_schema 将自动反映这一点。

提供的 shop.db 的已知数据特征： customers 没有 country 列，因此无法回答“来自德国的客户”这类问题——模式工具使这一点可被发现，而 query_database 会返回明确的 no such column: country 错误，而不是猜测。数据库中目前所有 750 个订单都日期在 2026 年（2025 年没有），因此“2025 年收入”查询会正确返回 0/null，而不是错误。

安装

cd mcp-sql
python3 -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install -r requirements.txt

配置

数据库路径在源代码中从未硬编码。它按以下方式解析：

1. 如果设置了 SHOP_DB_PATH 环境变量，则使用它；

2. 否则使用 server.py 旁边的 shop.db。

将 .env.example 复制为 .env，如果希望将服务器指向不同的数据库文件，请编辑它（你需要自己将其加载到 shell/代理启动器中，例如 export $(cat .env | xargs)，或者直接设置 SHOP_DB_PATH）：

cp .env.example .env
# edit .env, or simply:
export SHOP_DB_PATH=/absolute/path/to/shop.db

运行

source .venv/bin/activate
python server.py

该进程通过 stdio 使用 MCP 协议，并等待客户端——它看起来会“卡住”且没有输出，这是预期的：连接一个 MCP 客户端（AI 代理，或 mcp-inspector，见下文），而不是在终端中独立运行它。

使用官方 MCP Inspector 进行快速手动检查（无需安装）：

npx @modelcontextprotocol/inspector --cli .venv/bin/python server.py --method tools/list

连接到 AI 代理

大多数兼容 MCP 的客户端（Claude Desktop、Claude Code 等）读取类似 mcp-config.example.json 的 JSON 配置块：

{
  "mcpServers": {
    "shop-mcp": {
      "command": "/absolute/path/to/mcp-sql/.venv/bin/python",
      "args": ["/absolute/path/to/mcp-sql/server.py"],
      "env": {
        "SHOP_DB_PATH": "/absolute/path/to/mcp-sql/shop.db"
      }
    }
  }
}

注意：

• 使用 venv 的 Python 解释器的绝对路径（如上所示），以便无需手动激活 venv 就能找到 mcp 包；如果 mcp 安装在解析到的任何环境中，使用裸 python3 也可以。

• SHOP_DB_PATH 是可选的——省略它则使用捆绑的 shop.db。

• 绝对路径属于此配置文件，由连接服务器的任何人提供——绝不在 server.py 本身内部。

• 此块在客户端中的具体位置因客户端而异（例如，Claude Desktop 使用 claude_desktop_config.json，具有相同的 mcpServers 结构；其他客户端可能只需要内部的 {"command": ..., "args": ..., "env": ...} 对象）。请查阅客户端的文档以了解文件位置。

测试

source .venv/bin/activate
python -m pytest tests/ -v

这运行 48 个测试，包括：

• 模式发现（表、列、主键/外键、关系、行数）；

• SELECT、JOIN、WHERE、GROUP BY、ORDER BY、聚合函数（COUNT/SUM/AVG/MIN/MAX）、子查询、安全的 WITH ... SELECT CTE，以及日期过滤（strftime）；

• 行数限制钳制和基于偏移量的分页；

• 对无效 SQL、未知表/列、空查询和缺失数据库文件的友好错误处理；

• 只读安全性：任务中列出的每种语句类型（DELETE、UPDATE、DROP、CREATE、INSERT，以及 ALTER、REPLACE、TRUNCATE、ATTACH、DETACH、VACUUM、REINDEX、破坏性的 PRAGMA、堆叠的 SELECT 1; DROP TABLE ...，以及 WITH x AS (...) DELETE ... CTE 伪装的删除）都被拒绝，并且之后断言数据库文件的行数和 SHA-256 哈希不变；

• tests/test_stdio_integration.py 将 server.py 作为真正的子进程启动，并通过实际的 MCP 客户端 SDK 通过 stdio 驱动它（initialize → list_tools → call_tool），而不是直接调用 Python 函数——这与真实代理使用的路径相同。

MCP 工具

get_database_schema()

无参数。当你尚不知道确切的表/列名时，请先调用此函数——不要猜测。每个表返回：row_count、columns（名称、SQLite 类型、not_null、default_value、is_primary_key）、primary_key、foreign_keys（列、引用的表/列、ON DELETE/ON UPDATE），以及一些 sample_rows，以便代理可以看到真实的日期格式、状态值、价格量级等。顶层的 relationships 列表给出从实时外键派生的 table.column -> other_table.column 字符串。

query_database(sql, limit=100, offset=0)

运行一条只读 SQL 语句（SELECT 或 WITH ... SELECT），并返回 {columns, rows, row_count, limit, offset, truncated, total_matching_rows}。支持 JOIN、WHERE、GROUP BY、ORDER BY、聚合函数、子查询和 CTE。limit 被钳制在 1..500（默认 100）；使用 offset 分页浏览更大的结果。total_matching_rows 和 truncated 告诉调用者当前页是完整结果还是还有更多要获取。错误（语法错误、未知表/列或拒绝的写入尝试）以简短、具体的消息抛出——绝不是原始的 Python 回溯。

安全性：如何强制只读

任务明确要求不要依赖单一的正则表达式/关键字检查，因此此服务器分层了四个独立的防御措施——在 tests/test_server.py 中验证：

1. 操作系统级别的只读文件句柄。 SQLite 文件以 URI file:<path>?mode=ro 打开。SQLite 本身随后会拒绝任何写入（OperationalError: attempt to write a readonly database），无论执行什么 SQL——即使下面的每个检查都有 bug，这也成立。

2. PRAGMA query_only = ON 在每个连接上设置，作为第二个独立的 SQLite 级别写入防护。

3. sqlite3 授权器回调（Connection.set_authorizer）在 SQLite 引擎级别仅允许 SELECT / READ / FUNCTION / RECURSIVE 操作，并拒绝其他所有操作——INSERT、UPDATE、DELETE、DROP、ALTER、CREATE、REPLACE、TRUNCATE、ATTACH、DETACH、VACUUM、REINDEX、PRAGMA、事务等。这运行在解析后的语句上，因此它也能捕获经典的 CTE 绕过 WITH x AS (SELECT 1) DELETE FROM ...，而天真的“必须以 SELECT 开头”的文本检查会漏掉这一点。

4. server.py 中的语句形状检查：提交的文本必须以 SELECT/WITH 开头（在接触 SQLite 之前快速、友好地拒绝），并且每个查询都包装为 SELECT * FROM (<query>) LIMIT :limit OFFSET :offset 执行——单个语句是解析所必需的，因此堆叠的 SELECT 1; DROP TABLE customers 会变成普通的 SQL 语法错误，而不是执行两条语句。

因为第 1 层（mode=ro）由 SQLite/操作系统独立于服务器自身的逻辑强制执行，所以即使第 2-4 层存在 bug，shop.db 也无法通过此服务器被修改。

已知限制

• 提供的 shop.db 中 customers 没有 country/位置列，因此无法从这些数据回答“来自德国的客户”等问题——模式工具会暴露这一点，而不是服务器发明一个列。

• 提供的数据中所有订单都日期在 2026 年；2025 年收入查询正确返回 0 而不是错误。

• query_database 中的 total_matching_rows 通过第二个包装相同查询的 COUNT(*) 计算；对于非常昂贵的查询，这大约会使工作量翻倍。鉴于这个数据库的大小（每个表几百到几千行），这不是实际关注点。