---
slug: 45-db-mcp
title: Tu DB como tools MCP
description: Exponé la misma base whitelisteada y filtrada por RLS a agentes de IA como tools MCP — una task JSON-RPC pura, con el user pasado explícito en cada llamada.
example_ids: [db-mcp]
---

# Tu DB como tools MCP

**[Servers MCP](/es/0.6.x/42-mcp)** muestra el protocolo. Esta página lo aplica a tus datos: la **misma** declaración de whitelist + row-level security que alimenta **[tu API REST](/es/0.6.x/44-db-api)** se vuelve un set de tools MCP que un agente (Claude, Cursor, cualquier cliente) puede llamar — y RLS filtra las filas para el user del agente **idéntico** a REST.

```synsema
-- Doc example: the same DB exposed as MCP tools — one pure task speaks JSON-RPC.
-- REST and MCP share ONE userland "declaration" (whitelist + RLS): the mcp() task here
-- is what the page (45-db-mcp) wires to `route "POST /mcp"`. The `user` is threaded
-- EXPLICITLY into every tool call — RLS works identically over MCP.
intent: "doc example: DB-as-MCP — data tools with row-level security"
require db(":memory:")

db_open(":memory:")
sql_exec("CREATE TABLE carts (id INTEGER PRIMARY KEY, customer_phone TEXT, status TEXT)")
sql_exec("INSERT INTO carts (customer_phone, status) VALUES ('+549115555', 'open')")
sql_exec("INSERT INTO carts (customer_phone, status) VALUES ('+549119999', 'open')")

let EXPOSED be ["carts"]

task rls_allow(t, row, user)
    when t == "carts"
        give user.role == "admin" or row.customer_phone == user.phone
    give true

task table_rows(t, user)
    when not contains(EXPOSED, t)
        give nothing
    give where(sql("SELECT * FROM " + t), (r) => rls_allow(t, r, user))

-- One task = the whole MCP server (initialize / tools/list / tools/call).
task mcp(req, u)
    let m be req["method"]
    let rid be when contains(req, "id") then req["id"] otherwise nothing
    when m == "initialize"
        give {"jsonrpc": "2.0", "id": rid, "result": {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "my-db", "version": "0.1"}}}
    when m == "tools/list"
        give {"jsonrpc": "2.0", "id": rid, "result": {"tools": [
            {"name": "table_rows", "description": "Rows of an exposed table, filtered by row-level security.", "inputSchema": {"type": "object", "properties": {"table": {"type": "string"}}, "required": ["table"]}}
        ]}}
    when m == "tools/call"
        let p be req["params"]
        when p["name"] == "table_rows"
            let rows be table_rows(p["arguments"]["table"], u)
            when rows == nothing
                give {"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": "no such resource"}}
            give {"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": json_encode(rows)}]}}
        give {"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": "unknown tool"}}
    give {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": "method not found"}}

let ADMIN be {"user": "root", "role": "admin", "phone": ""}
let MARIA be {"user": "maria", "role": "customer", "phone": "+549115555"}

print("tools: " + text(length(mcp({"method": "tools/list", "id": 1}, ADMIN)["result"]["tools"])))

test "initialize answers the MCP handshake"
    let r be mcp({"method": "initialize", "id": 1}, ADMIN)
    assert_eq(r["result"]["protocolVersion"], "2024-11-05")

test "tools/call threads the user: RLS filters rows identically to REST"
    let all be mcp({"method": "tools/call", "id": 2, "params": {"name": "table_rows", "arguments": {"table": "carts"}}}, ADMIN)
    assert_eq(length(json_decode(all["result"]["content"][0]["text"])), 2)
    let mine be mcp({"method": "tools/call", "id": 3, "params": {"name": "table_rows", "arguments": {"table": "carts"}}}, MARIA)
    assert_eq(length(json_decode(mine["result"]["content"][0]["text"])), 1)

test "unknown tool and unknown method are JSON-RPC errors, not crashes"
    let e1 be mcp({"method": "tools/call", "id": 4, "params": {"name": "nope", "arguments": {}}}, ADMIN)
    assert_eq(e1["error"]["code"], 0 - 32602)
    let e2 be mcp({"method": "resources/list", "id": 5}, ADMIN)
    assert_eq(e2["error"]["code"], 0 - 32601)
```

## Una task pura es todo el server

`mcp(req, u)` maneja `initialize`, `tools/list` y `tools/call` y devuelve maps JSON-RPC. Por ser una task pura, se testea sin server — el wiring es una ruta:

```synsema
serve on 8080
    auth with check_token

    route "POST /mcp" requires auth
        give mcp(json of request, user of request)
```

## Pasá el `user` explícito

La línea crítica es `mcp(json of request, user of request)`: el user autenticado viaja **como parámetro** a cada llamada de tool, y cada tool lo baja a `table_rows(t, u)` — la misma puerta RLS de REST. No hay "usuario actual" ambiente que olvidar; si una tool no recibe `u`, no puede leer nada.

```synsema
when p["name"] == "table_rows"
    let rows be table_rows(p["arguments"]["table"], u)   -- RLS filtra para ESTE user
```

Un cliente llamando `table_rows` por MCP ve exactamente las filas que vería por REST — una declaración, dos protocolos.

## Semántica de fallas

Tool desconocida → error JSON-RPC `-32602`; método desconocido → `-32601`; tabla no expuesta → un error como resultado, no un crash. Los agentes leen los errores: mantenelos estructurados y autocontenidos.

> Aplica la misma advertencia honesta de la página REST: RLS es **disciplina** — toda tool debe leer por la puerta. La ventaja de la declaración compartida es que hay exactamente una puerta que revisar.