Northwind MCP

Northwind MCP服务器

      

概述

此项目是安全模型上下文协议（MCP）服务器的参考实现，该服务器为LLM代理提供了对经典Northwind SQLite3数据库的安全、只读SQL访问。

它展示了多层SQL验证、模式感知护栏和防御性AI与数据库交互。

特性

• MCP工具接口-通过流式HTTP为现代LLM代理公开工具。
• 安全的SQL执行-通过多层验证强制执行仅SELECT查询。
• 模式感知护栏–防止无效或产生幻觉的表/列。
• 飞行前验证-使用SQLite EXPLAIN在执行前验证查询。
• 只读数据库模式-保证没有数据突变。
• 分层设计——工具处理、验证和执行之间的明确分离。
• 综合测试——单元和集成测试，以确保核心逻辑的覆盖。

暴露的MCP工具

技术栈

• 主控程序 （通过FastMCP）：定义并公开用于通过Streamable HTTP进行LLM交互的结构化工具。
• SQLite3 （只读模式）：用于演示安全、不可变查询执行的轻量级数据库。
• sqlparse：解析和验证SQL结构。确保：单语句，仅SELECT。
• Pydantic v2：为工具输出提供严格的类型化响应模型。
• 紫外线：使用确定性锁文件管理依赖关系，以实现可复制的构建。
• Pytest （+任何IO）：用于具有异步支持和覆盖率报告的单元和集成测试。

建筑

系统上下文

下面的流程图显示了Northwind MCP服务器如何融入更大的生态系统。

flowchart LR
    User[LLM Agent / AI Assistant]
    MCP[Northwind MCP Server]
    DB[(Northwind SQLite Database)]

    User -->|MCP over Streamable HTTP| MCP
    MCP -->|Read-only SQL| DB

描述

• LLM代理通过模型上下文协议（流式HTTP传输）进行通信。
• MCP服务器充当安全边界。
• SQLite数据库从不直接暴露给代理。

这加强了人工智能推理和数据执行之间的适当隔离。

逻辑概述

服务器位于MCP兼容代理和SQLite数据库之间，在执行之前强制执行严格的验证。

flowchart TD
    A[LLM Agent / MCP Client] -->|Streamable HTTP| B[Northwind MCP Server]

    B --> C[get_db_schema]
    B --> D[validate_query]
    B --> E[execute_sql]

    D --> F[SQL Guardrails]
    F --> G[Schema Validation]
    F --> H[Keyword Checks]
    F --> I[SQLite EXPLAIN]

    E --> J[(Northwind SQLite DB)]

    C --> J

项目结构

northwind-mcp/
│
├── pyproject.toml              # Project configuration
├── uv.lock                     # Deterministic dependency lockfile
├── requirements.txt            # Runtime dependencies
├── requirements-dev.txt        # Development dependencies
├── .python-version             # Python version pin
├── README.md                   # Project documentation
│
├── northwind_mcp/              # Application package
│   ├── __init__.py
│   ├── main.py                 # ASGI entrypoint (Streamable HTTP app)
│   ├── server.py               # MCP tool definitions
│   ├── connection.py           # SQLite3 connection handling
│   ├── logging_config.py       # Logging configuration
│   │
│   ├── models/
│   │   └── schema.py           # Pydantic response models
│   │
│   ├── utils/
│   │   └── utils.py            # Utility functions
│   │
│   └── data/                   # SQLite3 DB file (Northwind DB to be copied here)
│
└── tests/
    ├── conftest.py             # Shared test fixtures
    ├── unit/                   # Unit test cases
    └── integration/            # Integration test cases

安装说明

1.克隆存储库

克隆后，切换到终端中的项目根目录。

cd northwind-mcp

2.安装依赖项

选项1——使用 uv （推荐）

uv sync

对于开发依赖关系：

uv sync --group dev

选项2——使用 pip

pip install -r requirements.txt
pip install -r requirements-dev.txt  # For development dependencies

3.下载Northwind SQLite数据库

数据库文件被有意排除在版本控制之外，以保持存储库的轻量级。

从以下网址下载：https://github.com/jpwhite3/northwind-SQLite3.

下载后，复制 northwind.db 将文件放入目录：

northwind_mcp/data/

跟随 curl 命令可用于下载和保存db文件：

curl -fsSL -o northwind_mcp/data/northwind.db \
https://raw.githubusercontent.com/jpwhite3/northwind-SQLite3/main/dist/northwind.db

4.运行MCP服务器

注： 日志级别配置可以通过环境变量进行控制 LOG_LEVEL （默认为 INFO 如果没有提供）。

选项1-Python模块

export LOG_LEVEL=WARNING
python -m northwind_mcp.main

选项2-Uvicorn直接

export LOG_LEVEL=DEBUG
uvicorn northwind_mcp.main:app --host 127.0.0.0 --port 9001

MCP服务器将在以下平台上运行： http://localhost:9001

Streamable HTTP端点将在以下位置可用：

http://localhost:9001/mcp

*（注意：根据FastMCP配置，服务器也可能直接在根节点响应/）*

运行测试

单元测试：

使用紫外线运行：

uv run pytest -m unit

或

pytest -m unit

覆盖率报告将自动生成。

Sample Coverage Output (100%)

uv run pytest -m unit

==============================================================
test session starts
==============================================================
platform darwin -- Python 3.12.1, pytest-9.0.2, pluggy-1.6.0
rootdir: /Users/....../....../northwind-mcp
configfile: pyproject.toml
testpaths: tests
plugins: anyio-4.12.1, mock-3.15.1, cov-7.0.0
collected 33 items / 15 deselected / 18 selected                                                                                     

tests/unit/test_connection.py ...
tests/unit/test_execute_sql.py ...
tests/unit/test_get_db_schema.py ..
tests/unit/test_validate_sql.py ..........

==============================================================
tests coverage
==============================================================
__________________________________________________
coverage: platform darwin, python 3.12.1-final-0
__________________________________________________

Name                             Stmts   Miss  Cover   Missing
--------------------------------------------------------------
northwind_mcp/__init__.py            0      0   100%
northwind_mcp/connection.py          8      0   100%
northwind_mcp/models/schema.py      13      0   100%
northwind_mcp/server.py             97      0   100%
northwind_mcp/utils/utils.py         9      0   100%
--------------------------------------------------------------
TOTAL                              127      0   100%
=================================================
18 passed, 15 deselected in 0.17s
==================================================

集成测试（需要运行服务器）：

使用紫外线运行：

uv run pytest -m integration

或

pytest -m integration

或 使用紫外线运行： uv run pytest -m unit

使用MCP检查器

官方MCP检查器UI可用于与Northwind MCP服务器交互。

启动MCP服务器

从项目根：

uv run python -m northwind_mcp.main

默认情况下，服务器运行在：

http://localhost:9001/mcp

启动MCP检查器

在单独的终端中：

npx @modelcontextprotocol/inspector

应看到与以下类似的启动输出：

Starting MCP inspector...
Proxy server listening on localhost:6277
Session token: 

MCP Inspector is up and running at:
http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=

打开MCP检查器UI

• 在浏览器中打开MCP检查器URL
• 将传输类型设置为： Streamable HTTP
• 输入MCP服务器URL： http://localhost:9001/mcp
• 点击连接
• 连接到 NorthwindMCP 应该建立
• 首选 Tools 节和 List Tools
• 应列出工具
• 通过提供输入和验证输出继续测试工具

MCP Inspector UI

MCP集成

此服务器与以下设备兼容：

• MCP兼容LLM代理
• FastMCP客户端
• 任何支持流式HTTP传输的MCP客户端

LLM说明

此服务器旨在让LLM遵循“验证然后执行”模式：

1. 发现：致电 get_db_schema 了解数据库模式。
2. 验证：呼叫 validate_query 检查语法或安全错误。
3. 执行：调用 execute_sql 只有在验证返回后 valid: true.