结构化数据MCP
SQLite的数据驱动MCP服务器，将数据库查询作为Claude Desktop工具公开，而无需编写Python代码。在YAML中定义查询，获取即时MCP工具。
特性
- 零代码扩展性：通过创建YAML文件添加新的数据库查询——不需要Python
- 类型安全：参数和约束的自动Pydantic验证
- 只读安全：通过SQL验证强制只读数据库访问
- 丰富的关系查询：支持多级JOIN和复杂聚合
- 分页支持：用于大型结果集的内置限制/偏移参数
- FastMCP集成：利用FastMCP实现无缝的Claude Desktop集成
建筑
YAML Query Definitions → QueryLoader → Pydantic Models → ToolFactory → MCP Tools
↓
QueryExecutor ← SQLValidator
↓
SQLite Database关键的原则：整个查询目录都存在于YAML文件中。Python层自动处理验证、执行和MCP工具生成。
快速开始
安装
# Clone the repository
git clone
cd StructuredDataMCP
# Create virtual environment
python -m venv .venv
source .venv/bin/activate
# Windows CMD: .venv\Scripts\activate
# PowerShell:
# Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser
# .\.venv\Scripts\Activate.ps1
# Install dependencies
pip install -e .设置示例数据库
# Create sample database with schema or use the sample.db in the repo as-is
sqlite3 examples/sample.db = datetime('now', '-' || :days || ' days')
ORDER BY o.created_at DESC
LIMIT :limit OFFSET :offset
parameters:
- name: "days"
type: "integer"
description: "Number of days to look back"
required: false
default: 7
constraints:
minimum: 1
maximum: 365
- name: "limit"
type: "integer"
description: "Maximum number of results"
required: false
default: 20
constraints:
minimum: 1
maximum: 100
- name: "offset"
type: "integer"
description: "Number of results to skip"
required: false
default: 0
constraints:
minimum: 0
output_schema:
type: "object"
properties:
results:
type: "array"
items:
type: "object"
properties:
order_id:
type: "integer"
username:
type: "string"
total_amount:
type: "number"
created_at:
type: "string"
count:
type: "integer"步骤2：重新启动MCP服务器
就是这样！不需要更改Python代码。服务器自动执行以下操作：
- 发现新的YAML文件
- 验证查询定义
- 生成具有类型安全参数的MCP工具
- 使其可用于Claude Desktop
查询定义模式
遵循以下约定以保持一致性：
SQL模式：
- 使用
COALESCE(field, default)在所有可为null的字段上 - 字符串默认为
''，数字到0 - 使用命名参数：
:param_name - 包括明确
ORDER BY条款 - 使用表别名（u、o、oi、p、c、r、ci）
分页：
- 多记录查询应包括
limit和offset参数 - 使用
LIMIT :limit OFFSET :offset在SQL中 - 设置合理的默认值（限制：10-20，偏移：0）
- 设置最大约束（限制：100）
输出架构：
- 始终将结果打包
{results: [...], count: N}格式 - 定义输出模式中的所有字段
- 将SQL列名与输出属性名匹配
命名：
- 查询名称：
{verb}_{entity}[_{modifier}]
- 示例： get_user_by_id, search_products, get_low_stock_products
- 参数名称：snake_case
- 使用描述性标题和描述
运作原理
1.查询加载(loader.py)
loader = QueryLoader(queries_dir="config/queries")
queries = loader.load_all_queries()全部加载 .yaml 文件，根据Pydantic模型进行验证(QueryFile, QueryDefinition).
2.工具生成(factory.py)
tool = ToolFactory.create_tool(query_def, executor)生成具有以下功能的FastMCP工具：
- 基于参数的动态函数签名
- 类型注解
- 默认值
- 完整文档字符串
3.查询执行(executor.py)
result = await executor.execute_query(query_def, arguments)强制执行：
- 只读数据库访问(
mode=ro) - SQL验证（无插入/更新/删除/删除）
- 参数验证（类型、约束）
- 结果限制（最多1000行）
- 查询超时（30秒）
4.MCP集成(main.py)
mcp = FastMCP("StructuredDataMCP")
@mcp.tool()
async def get_user_by_id(user_id: int):
# Auto-generated from YAML
...FastMCP处理MCP协议、Claude Desktop通信和工具调用。
安全功能
- 只读访问：数据库已在中打开
mode=ro - SQL验证：基于正则表达式的验证块写入操作
- 参数验证：类型检查和约束执行
- 结果限制：每个查询最多1000行
- 查询超时：30秒执行限制
- 无动态SQL：所有在YAML中预定义的查询
发展
项目结构
StructuredDataMCP/
├── config/
│ └── queries/ # YAML query definitions
│ ├── users.yaml
│ ├── products.yaml
│ ├── orders.yaml
│ ├── reviews.yaml
│ ├── cart.yaml
│ └── analytics.yaml
├── src/structureddatamcp/
│ ├── models.py # Pydantic models for YAML validation
│ ├── loader.py # YAML query loader
│ ├── validator.py # SQL validation
│ ├── executor.py # Query execution engine
│ └── factory.py # MCP tool factory
├── examples/
│ ├── schema_migration.sql # Database schema
│ ├── sample_data.sql # Sample data
│ └── sample.db # SQLite database
├── main.py # MCP server entry point
└── README.md运行测试
# Verify queries load correctly
source .venv/bin/activate
python main.py
# Should output:
# ✓ Loaded 19 query tools from .../config/queries直接测试查询
# Test a query against the database
sqlite3 examples/sample.db "
SELECT
COALESCE(u.username, '') as username,
COALESCE(COUNT(o.order_id), 0) as order_count
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id
GROUP BY u.user_id, u.username
ORDER BY order_count DESC;
"Claude Desktop使用示例
配置后，您可以向Claude提出与您的工具相对应的自然语言问题：
用户：“显示用户alice的所有订单”
克劳德: *用途 get_user_orders 使用搜索工具查找alice的user_id，然后检索订单*
用户“最畅销的三大产品是什么？”
克劳德: *用途 get_popular_products 限制工具=3*
用户：“哪些用户购买了无线耳机？”
克劳德: *用途 get_users_by_product 工具（反向遍历）*
用户：“显示笔记本电脑的评论统计数据”
克劳德: *用途 get_product_review_stats 产品搜索工具*
扩展到您自己的数据库
- 替换数据库:
cp your_database.db examples/sample.db- 探索模式:
sqlite3 examples/sample.db ".schema"- 创建查询定义:
- 在中添加YAML文件 config/queries/ - 遵循现有查询中的模式 - 使用COALESCE实现零安全 - 为多记录结果添加分页
- 测试查询:
# Test SQL directly first
sqlite3 examples/sample.db "YOUR SQL HERE"- 重新启动克劳德桌面
最佳实践
- 始终使用COALESCE：防止NULL值破坏JSON序列化
- 直接测试查询：在添加到YAML之前使用sqlite3 CLI
- 添加分页：多记录查询应支持限额/偏移
- 彻底记录：清晰的描述有助于克劳德选择合适的工具
- 使用约束：最小/最大值可防止滥用和错误
- 遵循命名约定：一致的命名提高了可发现性
- 从SELECT开始：根据设计，查询是只读的
故障排除
服务器未显示在Claude桌面中
- 检查配置文件语法（有效的JSON）
- 验证Python和main.py的绝对路径
- 检查克劳德桌面日志：
- macOS： ~/Library/Logs/Claude/ - 窗户： %APPDATA%\Claude\logs\
- 完全重新启动Claude Desktop（退出并重新启动）
查询不起作用
- 直接用sqlite3测试SQL
- 检查YAML语法（缩进、冒号）
- 验证所有可空字段上的COALESCE
- 检查参数名称是否与SQL匹配(
:param_name) - 查看服务器stderr输出是否存在验证错误
未找到数据库
确保路径 main.py 是绝对的还是相对于脚本位置的：
DB_PATH = Path(__file__).parent / "examples" / "sample.db"
QUERIES_DIR = Path(__file__).parent / "config" / "queries"许可证
麻省理工学院
贡献
欢迎投稿！拜托：
- 遵循现有的YAML模式
- 添加全面的描述
- 在PR描述中包含示例查询
- 提交前使用Claude Desktop进行测试
致谢
建于 FastMCP -构建MCP服务器的最快方法。