Servidor MCP SQLite (solo lectura)

Un servidor Model Context Protocol listo para producción que proporciona a los agentes de IA un acceso seguro y solo de lectura a una base de datos SQLite (shop.db). Construido con el SDK oficial mcp de Python mediante el transporte stdio.

Características

• 3 herramientas MCP: list_tables, describe_table, query_database

• Seguridad de solo lectura con defensa en profundidad: modo de solo lectura por URI de SQLite + PRAGMA query_only + validador SQL + inspección de opcodes de EXPLAIN

• Validación de consultas: rechaza INSERT/UPDATE/DELETE/DROP/ALTER/CREATE/REPLACE/TRUNCATE/ATTACH/DETACH, consultas de múltiples sentencias (;), comentarios SQL (--, /* */) y PRAGMA de modificación — sin falsos positivos en literales de cadena

• Paginación: límite de filas por defecto (100), parámetros limit/offset, indicador de salida truncada

• Registro solo en stderr: todos los registros y tracebacks van a sys.stderr; stdout está reservado exclusivamente para JSON-RPC

• Tipado completo: mypy --strict sin errores

• TDD: 105 pruebas que cubren seguridad, capa de base de datos, herramientas MCP, 8 consultas de referencia y la protección de stderr

Related MCP server: sqlite-mcp-server

Inicio rápido

Requisitos previos

• Python 3.10+

• Un archivo de base de datos SQLite (por defecto: ./shop.db)

Configuración local

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Configurar

Copia .env.example y establece la ruta de la base de datos:

cp .env.example .env
# Edit DATABASE_PATH to point to your SQLite file

O establece la variable de entorno directamente:

export DATABASE_PATH=/abs/path/to/shop.db

Ejecutar el servidor

python -m mcp_server.server

El servidor se comunica a través de stdin/stdout usando el transporte stdio de MCP. No interactúas con él directamente: un cliente MCP (p. ej., Claude Desktop, tu agente de IA) se conecta a él.

Configuraciones del cliente MCP

Python estándar

Añade esto a la configuración de tu cliente MCP (p. ej., el claude_desktop_config.json de Claude Desktop):

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "env": {
        "DATABASE_PATH": "/abs/path/to/shop.db"
      }
    }
  }
}

Docker

Primero construye la imagen:

docker build -t mcp-shop:latest .

A continuación, configura tu cliente MCP:

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-v", "/abs/path/to/shop.db:/app/shop.db",
        "-e", "DATABASE_PATH=/app/shop.db",
        "mcp-shop:latest"
      ]
    }
  }
}

Docker Compose

docker compose up -d

Herramientas

list_tables

Enumera todas las tablas y vistas de usuario de la base de datos (excluye las tablas internas sqlite_*).

Parámetros: ninguno

Devuelve:

{
  "tables": ["customers", "orders", "order_items", "products"],
  "count": 4
}

describe_table

Describe el esquema de una tabla: columnas, claves foráneas, número de filas y la sentencia CREATE.

Parámetros:

• table (cadena, obligatorio): nombre de la tabla a describir.

Devuelve:

{
  "table": "customers",
  "columns": [
    {"cid": 0, "name": "id", "type": "INTEGER", "notnull": 0, "default": null, "pk": 1},
    {"cid": 1, "name": "first_name", "type": "TEXT", "notnull": 1, "default": null, "pk": 0}
  ],
  "foreign_keys": [],
  "row_count": 150,
  "sql": "CREATE TABLE customers (...)"
}

query_database

Ejecuta una consulta SQL de solo lectura con soporte de paginación.

Parámetros:

• sql (cadena, obligatorio): una única sentencia SQL de solo lectura (SELECT, WITH, EXPLAIN o PRAGMA de solo lectura).

• limit (entero, opcional): número máximo de filas a devolver. Por defecto: 100. Máximo: 1000.

• offset (entero, opcional): número de filas a omitir. Por defecto: 0.

Devuelve:

{
  "columns": ["id", "first_name"],
  "rows": [{"id": 1, "first_name": "Alice"}, {"id": 2, "first_name": "Bob"}],
  "row_count": 2,
  "truncated": false,
  "limit": 100,
  "offset": 0
}

Cuando truncated es true, hay más filas disponibles: aumenta offset para obtener la siguiente página.

Seguridad

El servidor implementa defensa en profundidad para garantizar el acceso de solo lectura:

Capa 1: conexión SQLite (modo de solo lectura por URI)

La base de datos se abre con file:<path>?mode=ro, lo que impide escrituras a nivel del motor de SQLite. Además, se establece PRAGMA query_only = ON en cada conexión.

Capa 2: validador de consultas SQL (security.py)

Antes de que cualquier consulta llegue a SQLite, pasa por un validador de varias etapas:

1. Eliminación de literales de cadena: los literales de cadena ('...', "...") se sustituyen por marcadores de posición para que las palabras clave dentro de los datos (p. ej., un producto llamado «Deleted Item») no provoquen falsos positivos.

2. Detección de comentarios: los comentarios SQL (--, /* */) se rechazan para prevenir evasiones basadas en comentarios.

3. Rechazo de múltiples sentencias: se rechaza cualquier punto y coma (;), lo que impide consultas apiladas.

4. Análisis de palabras clave: la primera palabra clave real de la sentencia debe ser SELECT, WITH, EXPLAIN o PRAGMA. Se bloquean las palabras clave destructivas (INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, REPLACE, TRUNCATE, ATTACH, BLOQUEO, VACUUM, etc.).

5. Validación de PRAGMA: se permiten los PRAGMA de solo lectura (table_info, database_list, etc.). Se rechaza cualquier PRAGMA con asignación (=) o que esté en la lista de PRAGMA mutantes (journal_mode, synchronous, foreign_keys, etc.).

Capa 3: inspección de opcodes de EXPLAIN

Como defensa final, la consulta se procesa con el analizador de SQLite mediante EXPLAIN <query>. El flujo de opcodes resultante se inspecciona buscando opcodes de escritura (OpenWrite, Insert, Delete, Create, Drop, etc.) y los indicadores de transacción de escritura. Si se encuentra alguno, la consulta se rechaza.

Capa 4: mensajes de error saneados

Todos los errores devueltos al cliente se sanan: se eliminan las rutas del sistema de archivos y los detalles internos para evitar la fuga de información.

Pruebas

Las pruebas solo utilizan bases de datos temporales o en memoria — nunca la base de datos de producción shop.db.

# Run all tests
python -m pytest

# Run with verbose output
python -m pytest -v

# Run a specific test file
python -m pytest tests/test_security.py

Cobertura de pruebas

Análisis estático

# Type checking
python -m mypy

# Linting
python -m ruff check src/ tests/

Estructura del proyecto

.
├── .env.example          # Environment variable template
├── Dockerfile            # Docker containerization
├── docker-compose.yml    # Docker Compose config
├── pyproject.toml        # Package config, deps, tool settings
├── README.md             # This file
├── shop.db               # The SQLite database (not included in tests)
├── src/mcp_server/
│   ├── __init__.py
│   ├── config.py         # Configuration (DATABASE_PATH, limits, URI builder)
│   ├── db.py             # Read-only Database class with introspection + query
│   ├── security.py       # SQL validator (multi-layer defense-in-depth)
│   ├── server.py         # MCP server entrypoint (stdio transport)
│   ├── tools.py          # MCP tool definitions and handlers
│   └── py.typed          # PEP 561 marker
└── tests/
    ├── __init__.py
    ├── test_db.py        # Database layer + benchmark tests
    ├── test_security.py  # Query validator tests
    └── test_server.py    # MCP server/tool tests

Tareas de referencia

Las herramientas del servidor permiten a un agente de IA realizar estas tareas analíticas (validadas mediante pruebas sin con una base de datos de pruebas controlada):

1. Descubrimiento de tablas: list_tables + describe_table — listar todas las tablas y describir los esquemas.

2. Conteo filtrado: query_database con SELECT COUNT(*) FROM customers WHERE country = 'Germany'.

3. Agregación por país: SELECT country, COUNT(*) ... GROUP BY country ORDER BY ... DESC LIMIT 1.

4. LTV del cliente: unir customers + orders, SUM(total_amount), ordenar por total.

5. Rendimiento del producto: unir order_items + products, agregar por cantidad e ingresos, LIMIT 5.

6. Agregación por categoría: recorrer order_items → products → category, agregar ingresos, LIMIT 3.

7. Filtrado por fecha: SUM(total_amount) WHERE substr(order_date, 1, 4) = '2025'.

8. Agregación de pedidos: unir customers + orders, COUNT(o.id), ordenar por número de pedidos.

Configuración

Licencia

Este proyecto se proporciona tal cual, como referencia.Demostración.