Servidor MCP de solo lectura

import sqlite3

conn = sqlite3.connect("tienda.db")
conn.executescript("""
DROP TABLE IF EXISTS clientes;
DROP TABLE IF EXISTS pedidos;
CREATE TABLE clientes (id INTEGER PRIMARY KEY, nombre TEXT, ciudad TEXT);
CREATE TABLE pedidos (id INTEGER PRIMARY KEY, cliente_id INTEGER REFERENCES clientes(id),
                      fecha TEXT, importe REAL, estado TEXT);
INSERT INTO clientes VALUES (1, 'Ana López', 'Valencia'), (2, 'Panadería Lola', 'Sevilla');
INSERT INTO pedidos VALUES (1, 1, '2026-10-01', 49.90, 'entregado'),
                           (2, 1, '2026-10-12', 120.00, 'enviado'),
                           (3, 2, '2026-10-15', 15.50, 'pendiente');
""")
conn.commit()
conn.close()
print("tienda.db creada")

import logging
import os
import sqlite3
import time

from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

DB_PATH = os.environ.get("DB_PATH", "tienda.db")
MAX_FILAS = 200
MAX_SEGUNDOS = 5

# en stdio, stdout es el canal del protocolo: los logs van a stderr
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("tienda-mcp")

mcp = MCPServer("tienda-solo-lectura")


def _solo_lectura(accion, *args):
    # tercera capa: SQLite pregunta antes de cada operación y solo permitimos leer
    permitidas = {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION}
    return sqlite3.SQLITE_OK if accion in permitidas else sqlite3.SQLITE_DENY


def conectar() -> sqlite3.Connection:
    # primera capa: el archivo se abre en modo solo lectura (mode=ro)
    conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
    # segunda capa: la conexión rechaza cualquier escritura
    conn.execute("PRAGMA query_only = ON")
    conn.set_authorizer(_solo_lectura)
    # corta consultas que tarden demasiado
    limite = time.monotonic() + MAX_SEGUNDOS
    conn.set_progress_handler(lambda: 1 if time.monotonic() > limite else 0, 10_000)
    return conn


@mcp.tool()
def listar_tablas() -> list[str]:
    """Lista las tablas de la base de datos de la tienda. Úsala primero para saber qué hay."""
    conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
    try:
        filas = conn.execute(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
        ).fetchall()
        return [f[0] for f in filas]
    finally:
        conn.close()


@mcp.tool()
def describir_tabla(tabla: str) -> list[dict]:
    """Devuelve las columnas de una tabla (nombre y tipo). Úsala antes de escribir una consulta."""
    if tabla not in listar_tablas():
        raise ToolError(f"La tabla '{tabla}' no existe. Usa listar_tablas para ver las disponibles.")
    conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
    try:
        columnas = conn.execute(f'PRAGMA table_info("{tabla}")').fetchall()
        return [{"columna": c[1], "tipo": c[2]} for c in columnas]
    finally:
        conn.close()


@mcp.tool()
def consultar(sql: str, limite: int = 50) -> dict:
    """Ejecuta UNA consulta SELECT de solo lectura sobre la base de datos de la tienda.

    Devuelve columnas y filas. Como máximo `limite` filas (1-200); si hay más, `truncado` es true
    y conviene filtrar o agregar (WHERE, GROUP BY, COUNT). Cualquier escritura se rechaza.
    """
    texto = sql.strip().rstrip(";")
    if not texto.lower().startswith(("select", "with")):
        raise ToolError("Solo se permiten consultas SELECT (o WITH ... SELECT).")
    limite = max(1, min(limite, MAX_FILAS))
    conn = conectar()
    try:
        cursor = conn.execute(texto)
        filas = cursor.fetchmany(limite + 1)
        columnas = [d[0] for d in cursor.description]
    except sqlite3.Error as e:
        raise ToolError(f"La consulta falló: {e}. Revisa nombres con describir_tabla.") from e
    finally:
        conn.close()
    log.info("consulta ok: %s filas", len(filas))
    return {
        "columnas": columnas,
        "filas": [list(f) for f in filas[:limite]],
        "truncado": len(filas) > limite,
    }


if __name__ == "__main__":
    mcp.run()  # transporte stdio por defecto

import asyncio
import sys

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

PARAMS = StdioServerParameters(command=sys.executable, args=["servidor.py"], env={"DB_PATH": "tienda.db"})


async def main():
    async with stdio_client(PARAMS) as (lectura, escritura):
        async with ClientSession(lectura, escritura) as sesion:
            await sesion.initialize()
            herramientas = await sesion.list_tools()
            print("Herramientas:", [h.name for h in herramientas.tools])

            ok = await sesion.call_tool("consultar", {"sql": "SELECT nombre, ciudad FROM clientes"})
            print("SELECT:", ok.content[0].text)

            for sql in ["DELETE FROM pedidos", "WITH x AS (SELECT 1) DELETE FROM pedidos",
                        "SELECT 1; DROP TABLE pedidos"]:
                malo = await sesion.call_tool("consultar", {"sql": sql})
                print("Rechazada:" if malo.is_error else "¡ACEPTADA!", malo.content[0].text[:90])


asyncio.run(main())

{
  "mcpServers": {
    "tienda": {
      "command": "python",
      "args": ["/ruta/absoluta/servidor.py"],
      "env": { "DB_PATH": "/ruta/absoluta/tienda.db" }
    }
  }
}

# salida de ejemplo
Herramientas: ['listar_tablas', 'describir_tabla', 'consultar']
SELECT: {"columnas": ["nombre", "ciudad"], "filas": [["Ana López", "Valencia"], ...], "truncado": false}
Rechazada: Error executing tool consultar: Solo se permiten consultas SELECT (o WITH ... SELECT).
Rechazada: Error executing tool consultar: La consulta falló: not authorized. ...
Rechazada: Error executing tool consultar: La consulta falló: You can only execute one statement at a time.