SQLite数据库查询工具

YAML Query Definitions → QueryLoader → Pydantic Models → ToolFactory → MCP Tools
                                                              ↓
                                              QueryExecutor ← SQLValidator
                                                              ↓
                                                      SQLite Database

# Clone the repository
git clone <repository-url>
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 < examples/schema_migration.sql

# Load sample data
sqlite3 examples/sample.db < examples/sample_data.sql

python -c "import sqlite3; conn = sqlite3.connect('examples/sample.db'); print(sqlite3.sqlite_version)"

{
  "mcpServers": {
    "structured-data": {
      "command": "FULL_PATH_TO_CLONED_REPO_FOLDER/.venv/Scripts/python.exe",
      "args": [
        "FULL_PATH_TO_CLONED_REPO_FOLDER/main.py"
      ],
      "env": {}
    }
  }
}

users (6 records)
  ├→ orders (15 records) → order_items (27 records) → products (14 records) → categories (4 records)
  ├→ reviews (18 records) → products
  └→ cart_items (8 records) → products

# config/queries/my_queries.yaml
version: "1.0"
metadata:
  category: "custom"
  tags:
    - "reporting"

queries:
  - name: "get_recent_orders"
    title: "Get Recent Orders"
    description: "Retrieves orders from the last N days"

    sql: |
      SELECT
        COALESCE(o.order_id, 0) as order_id,
        COALESCE(u.username, '') as username,
        COALESCE(o.total_amount, 0) as total_amount,
        COALESCE(o.created_at, '') as created_at
      FROM orders o
      INNER JOIN users u ON o.user_id = u.user_id
      WHERE o.created_at >= 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"

loader = QueryLoader(queries_dir="config/queries")
queries = loader.load_all_queries()

tool = ToolFactory.create_tool(query_def, executor)

result = await executor.execute_query(query_def, arguments)

mcp = FastMCP("StructuredDataMCP")

@mcp.tool()
async def get_user_by_id(user_id: int):
    # Auto-generated from 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;
"

DB_PATH = Path(__file__).parent / "examples" / "sample.db"
QUERIES_DIR = Path(__file__).parent / "config" / "queries"