Model Database Protocol

MDBP - Protocolo de Base de Datos para Modelos

Protocolo de acceso a datos basado en intenciones para sistemas de IA.

MDBP permite el acceso seguro a bases de datos para LLMs. En lugar de generar SQL crudo, los LLMs producen objetos de intención estructurados. MDBP valida estas intenciones contra un registro de esquemas, aplica políticas de acceso, construye consultas parametrizadas mediante SQLAlchemy y devuelve respuestas amigables para LLMs.

LLM Intent (JSON) -> Schema Validation -> Policy Check -> SQLAlchemy Query -> Response

---

Tabla de Contenidos

• Instalación
• Inicio Rápido
• Conexión a Base de Datos Sin Configuración
• Conceptos Clave
• Tipos de Intención
• Filtrado
• Operaciones JOIN
• Agregación
• Campos Calculados
• CTE (Expresiones de Tabla Común)
• Operaciones de Escritura
• Operaciones de Conjuntos
• Registro de Esquemas
• Motor de Políticas
• Enmascaramiento de Datos
• Modo de Prueba (Dry-Run)
• Servidor MCP
• Manejo de Errores
• Referencia de API
• Seguridad

---

Instalación

pip install mdbp

Para desarrollo:

pip install mdbp[dev]

Requisitos:

• Python >= 3.10
• SQLAlchemy >= 2.0
• Pydantic >= 2.0
• mcp >= 1.0

Bases de Datos Soportadas: Cualquier backend compatible con SQLAlchemy: PostgreSQL, MySQL, SQLite, MSSQL, Oracle, BigQuery, etc.

---

Inicio Rápido

En Funcionamiento en 3 Líneas

from mdbp import MDBP

mdbp = MDBP(db_url="sqlite:///my.db")
result = mdbp.query({"intent": "list", "entity": "product", "limit": 10})

Cuando se llama a MDBP(db_url=...), todas las tablas y columnas se descubren automáticamente desde la base de datos. No se requiere registro manual.

Ejemplo de Salida

{
    "success": true,
    "intent": "list",
    "entity": "product",
    "summary": "10 product(s) found",
    "data": [
        {"id": 1, "name": "Laptop", "price": 15000},
        {"id": 2, "name": "Mouse", "price": 250}
    ]
}

Respuesta de Error

{
    "success": false,
    "intent": "list",
    "entity": "spaceship",
    "error": {
        "code": "MDBP_SCHEMA_ENTITY_NOT_FOUND",
        "message": "Entity 'spaceship' not found in schema registry.",
        "details": {
            "entity": "spaceship",
            "available_entities": ["product", "order", "customer"]
        }
    }
}

Cuando un LLM alucina un nombre de tabla, MDBP lo detecta y devuelve la lista de entidades disponibles. El LLM puede autocorregirse usando esta retroalimentación.

Ejemplo del Mundo Real (PostgreSQL)

from mdbp import MDBP

mdbp = MDBP(
    db_url="postgresql+psycopg2://user:password@localhost:5432/mydb",
    allowed_intents=["list", "get", "count", "aggregate"],  # read-only mode
)

# Auto-discovers all tables and columns
schema = mdbp.describe_schema()
for entity, info in schema.items():
    print(f"{entity}: {len(info['fields'])} fields")

# List with sorting and limit
result = mdbp.query({
    "intent": "list",
    "entity": "stock_price",
    "fields": ["Date", "Close", "Volume"],
    "sort": [{"field": "Date", "order": "desc"}],
    "limit": 5,
})
for row in result["data"]:
    print(f"{row['Date']} | ${row['Close']:.2f} | Vol: {row['Volume']:,}")

# Aggregation
result = mdbp.query({
    "intent": "aggregate",
    "entity": "stock_price",
    "aggregation": {"op": "avg", "field": "Close"},
})
print(f"Average close: ${float(result['data'][0]['result']):.2f}")

# Count with filters
result = mdbp.query({
    "intent": "count",
    "entity": "stock_price",
    "filters": {"Close__gte": 100},
})
print(f"Days above $100: {result['data']['count']}")

# Hallucination protection
result = mdbp.query({"intent": "list", "entity": "nonexistent_table"})
print(result["error"]["code"])           # MDBP_SCHEMA_ENTITY_NOT_FOUND
print(result["error"]["details"])        # {"available_entities": [...]}

mdbp.dispose()

---

Conexión a Base de Datos Sin Configuración

MDBP soporta cualquier base de datos compatible con SQLAlchemy sin código adicional. Solo instala el controlador y conéctate:

pip install mdbp
mdbp-server --db-url <DATABASE_URL>

Todas las tablas, columnas y tipos se descubren automáticamente — no se necesita definición de esquema ni código de servidor.

Ejemplos:

# SQLite
mdbp-server --db-url sqlite:///my.db

# PostgreSQL
pip install psycopg2
mdbp-server --db-url postgresql+psycopg2://user:pass@localhost/mydb

# MySQL
pip install pymysql
mdbp-server --db-url mysql+pymysql://user:pass@localhost/mydb

# BigQuery
pip install sqlalchemy-bigquery
gcloud auth application-default login
mdbp-server --db-url bigquery://project-id/dataset

# SQL Server
pip install pyodbc
mdbp-server --db-url mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server

Claude Desktop / Cursor / Clientes MCP:

{
    "mcpServers": {
        "my-database": {
            "command": "mdbp-server",
            "args": ["--db-url", "sqlite:///my.db"]
        }
    }
}

---

Conceptos Clave

¿Qué es una Intención?

Una intención es un objeto JSON estructurado que describe una operación de base de datos. Cada intención contiene estos campos principales:

Campo	Tipo	Requerido	Descripción
intent	string	Sí	Tipo de operación: list, get, count, aggregate, create, update, delete
entity	string	Sí	Nombre de la tabla/entidad objetivo
filters	object	No	Condiciones de filtro
fields	array	No	Campos a devolver (vacío = todos)
sort	array	No	Ordenamiento
limit	integer	No	Límite de resultados
offset	integer	No	Desplazamiento de paginación

Pipeline

Cada llamada a mdbp.query() pasa por estas etapas:

1. Parse       -> Convert dict to Intent model (Pydantic validation)
2. Whitelist   -> Check allowed_intents (global restriction)
3. Schema      -> Verify entity and fields exist in schema registry
4. Policy      -> Role-based access control, field restrictions
5. Plan        -> Convert Intent to SQLAlchemy statement
6. [Dry-run?]  -> Return compiled SQL without executing (if enabled)
7. Execute     -> Run parameterized query
8. Mask        -> Apply data masking to result fields (if configured)
9. Format      -> Convert result to LLM-friendly JSON

---

Tipos de Intención

list - Listar Registros

mdbp.query({
    "intent": "list",
    "entity": "product",
    "fields": ["name", "price"],
    "filters": {"price__gte": 100},
    "sort": [{"field": "price", "order": "desc"}],
    "limit": 10,
    "offset": 0,
    "distinct": True
})

get - Obtener un Solo Registro

mdbp.query({
    "intent": "get",
    "entity": "product",
    "id": 42
})

Devuelve un solo registro por clave primaria. Devuelve error MDBP_NOT_FOUND si no existe el registro.

count - Contar Registros

mdbp.query({
    "intent": "count",
    "entity": "product",
    "filters": {"category": "electronics"}
})

Salida:

{"success": true, "data": {"count": 156}}

aggregate - Agregar

mdbp.query({
    "intent": "aggregate",
    "entity": "order",
    "aggregation": {"op": "sum", "field": "amount"}
})

Operaciones soportadas: sum, avg, min, max, count

Múltiples agregaciones:

mdbp.query({
    "intent": "aggregate",
    "entity": "order",
    "aggregations": [
        {"op": "count", "field": "id"},
        {"op": "sum", "field": "amount"},
        {"op": "avg", "field": "amount"}
    ],
    "group_by": ["status"]
})

create - Crear Registro

mdbp.query({
    "intent": "create",
    "entity": "product",
    "data": {"name": "Laptop", "price": 999.99},
    "returning": ["id", "name"]
})

update - Actualizar Registro

mdbp.query({
    "intent": "update",
    "entity": "product",
    "id": 5,
    "data": {"price": 899.99}
})

Actualización masiva con filtros:

mdbp.query({
    "intent": "update",
    "entity": "product",
    "filters": {"status": "draft"},
    "data": {"status": "published"}
})

delete - Eliminar Registro

mdbp.query({
    "intent": "delete",
    "entity": "product",
    "id": 5
})

---

Filtrado

Filtros Simples (Sufijo de Operador)

Agrega un sufijo al nombre del campo en el diccionario filters para especificar el operador:

mdbp.query({
    "intent": "list",
    "entity": "product",
    "filters": {
        "category": "electronics",       # equality (=)
        "price__gt": 100,                # greater than (>)
        "price__lte": 5000,              # less than or equal (<=)
        "name__like": "%laptop%",        # LIKE
        "status__ne": "deleted",          # not equal (!=)
        "color__in": ["red", "blue"],     # IN (...)
        "stock__not_null": True,          # IS NOT NULL
    }
})

Todos los Operadores:

Sufijo	Equivalente SQL	Ejemplo
(ninguno)	=	{"city": "Istanbul"}
__gt	>	{"price__gt": 100}
__gte	>=	{"price__gte": 100}
__lt	<	{"price__lt": 500}
__lte	<=	{"price__lte": 500}
__ne	!=	{"status__ne": "deleted"}
__like	LIKE	{"name__like": "%phone%"}
__ilike	ILIKE	{"name__ilike": "%Phone%"}
__not_like	NOT LIKE	{"name__not_like": "%test%"}
__in	IN (...)	{"id__in": [1, 2, 3]}
__not_in	NOT IN	{"id__not_in": [4, 5]}
__between	BETWEEN	{"price__between": [100, 500]}
__null	IS NULL	{"email__null": true}
__not_null	IS NOT NULL	{"email__not_null": true}

Filtros Complejos (where)

Usa el campo where para lógica anidada AND/OR/NOT:

mdbp.query({
    "intent": "list",
    "entity": "product",
    "where": {
        "logic": "or",
        "conditions": [
            {"field": "category", "op": "eq", "value": "electronics"},
            {
                "logic": "and",
                "conditions": [
                    {"field": "price", "op": "lt", "value": 50},
                    {"field": "stock", "op": "gt", "value": 0}
                ]
            }
        ]
    }
})

Equivalente SQL:

WHERE category = 'electronics' OR (price < 50 AND stock > 0)

Ejemplo con NOT:

"where": {
    "logic": "not",
    "conditions": [
        {"field": "status", "op": "eq", "value": "deleted"}
    ]
}

Ejemplo con EXISTS:

"where": {
    "logic": "and",
    "conditions": [
        {
            "op": "exists",
            "subquery": {
                "intent": "list",
                "entity": "order",
                "fields": ["id"],
                "filters": {"customer_id": 1}
            }
        }
    ]
}

Filtros con Subconsultas

Usa $query en los valores de filtro para subconsultas:

mdbp.query({
    "intent": "list",
    "entity": "product",
    "filters": {
        "category_id__in": {
            "$query": {
                "intent": "list",
                "entity": "category",
                "fields": ["id"],
                "filters": {"name": "electronics"}
            }
        }
    }
})

Equivalente SQL:

SELECT * FROM products
WHERE category_id IN (SELECT id FROM categories WHERE name = 'electronics')

---

Operaciones JOIN

JOIN Básico

mdbp.query({
    "intent": "list",
    "entity": "order",
    "fields": ["product", "amount", "customer.name"],
    "join": [{
        "entity": "customer",
        "type": "inner",
        "on": {"customer_id": "id"}
    }]
})

• on: formato {local_field: foreign_field}
• Notación de punto en fields: "customer.name" se resuelve a la columna de la tabla unida
• type: inner, left, right, full

Múltiples JOINs

mdbp.query({
    "intent": "list",
    "entity": "order_item",
    "fields": ["quantity", "order.status", "product.name"],
    "join": [
        {"entity": "order", "type": "inner", "on": {"order_id": "id"}},
        {"entity": "product", "type": "inner", "on": {"product_id": "id"}}
    ]
})

Auto-JOIN (Alias)

mdbp.query({
    "intent": "list",
    "entity": "employee",
    "fields": ["name", "manager.name"],
    "join": [{
        "entity": "employee",
        "alias": "manager",
        "type": "left",
        "on": {"manager_id": "id"}
    }]
})

---

Agregación

GROUP BY

mdbp.query({
    "intent": "aggregate",
    "entity": "order",
    "aggregation": {"op": "count", "field": "id"},
    "group_by": ["status"]
})

HAVING

mdbp.query({
    "intent": "aggregate",
    "entity": "order",
    "aggregation": {"op": "sum", "field": "amount"},
    "group_by": ["customer_id"],
    "having": [{
        "op": "sum",
        "field": "amount",
        "condition": "gt",
        "value": 10000
    }]
})

SQL: HAVING SUM(amount) > 10000

Modos Avanzados de GROUP BY

# ROLLUP
mdbp.query({
    "intent": "aggregate",
    "entity": "sale",
    "aggregation": {"op": "sum", "field": "amount"},
    "group_by": ["year", "quarter"],
    "group_by_mode": "rollup"
})

# CUBE
mdbp.query({
    "intent": "aggregate",
    "entity": "sale",
    "aggregation": {"op": "sum", "field": "amount"},
    "group_by": ["region", "product"],
    "group_by_mode": "cube"
})

# GROUPING SETS
mdbp.query({
    "intent": "aggregate",
    "entity": "sale",
    "aggregation": {"op": "sum", "field": "amount"},
    "group_by": ["region", "product"],
    "group_by_mode": "grouping_sets",
    "grouping_sets": [["region"], ["product"], []]
})

---

Campos Calculados

CASE WHEN

mdbp.query({
    "intent": "list",
    "entity": "product",
    "fields": ["name", "price"],
    "computed_fields": [{
        "name": "price_tier",
        "case": {
            "when": [
                {"condition": {"field": "price", "op": "gt", "value": 1000}, "then": "premium"},
                {"condition": {"field": "price", "op": "gt", "value": 100}, "then": "standard"}
            ],
            "else_value": "budget"
        }
    }]
})

Funciones de Ventana

mdbp.query({
    "intent": "list",
    "entity": "product",
    "fields": ["name", "price", "category_id"],
    "computed_fields": [{
        "name": "price_rank",
        "window": {
            "function": "rank",
            "partition_by": ["category_id"],
            "order_by": [{"field": "price", "order": "desc"}]
        }
    }]
})

Funciones de ventana soportadas: rank, dense_rank, row_number, ntile, lag, lead, first_value, last_value, sum, avg, min, max, count

Funciones Escalares

mdbp.query({
    "intent": "list",
    "entity": "user",
    "fields": ["id"],
    "computed_fields": [
        {
            "name": "email_upper",
            "function": {"name": "upper", "args": ["email"]}
        },
        {
            "name": "display_name",
            "function": {
                "name": "coalesce",
                "args": ["nickname", {"literal": "Anonymous"}]
            }
        },
        {
            "name": "price_int",
            "function": {"name": "cast", "args": ["price"], "cast_to": "integer"}
        },
        {
            "name": "order_year",
            "function": {"name": "extract", "args": [{"literal": "year"}, "created_at"]}
        }
    ]
})

Funciones escalares soportadas: coalesce, upper, lower, cast, concat, trim, length, abs, round, substring, extract, now, current_date, replace

Nota: extract requiere el primer argumento como {"literal": "part"} donde la parte es year, month, day, hour, minute, o second.

---

CTE (Expresiones de Tabla Común)

mdbp.query({
    "intent": "list",
    "entity": "product",
    "fields": ["name", "price"],
    "cte": [{
        "name": "expensive_categories",
        "query": {
            "intent": "aggregate",
            "entity": "product",
            "aggregation": {"op": "avg", "field": "price"},
            "group_by": ["category_id"],
            "having": [{"op": "avg", "field": "price", "condition": "gt", "value": 500}]
        }
    }],
    "filters": {
        "category_id__in": {"$cte": "expensive_categories", "field": "category_id"}
    }
})

---

Operaciones de Escritura

batch_create - Inserción Masiva

mdbp.query({
    "intent": "batch_create",
    "entity": "product",
    "rows": [
        {"name": "Laptop", "price": 15000},
        {"name": "Mouse", "price": 250},
        {"name": "Keyboard", "price": 800}
    ]
})

upsert - Insertar o Actualizar

mdbp.query({
    "intent": "upsert",
    "entity": "product",
    "data": {"id": 1, "name": "Laptop Pro", "price": 18000},
    "conflict_target": ["id"],
    "conflict_update": ["name", "price"]
})

SQL: INSERT ... ON CONFLICT (id) DO UPDATE SET name=..., price=...

UPDATE con JOIN

mdbp.query({
    "intent": "update",
    "entity": "order",
    "data": {"status": "vip_order"},
    "from_entity": "customer",
    "from_join_on": {"customer_id": "id"},
    "from_filters": {"tier": "vip"}
})

RETURNING

mdbp.query({
    "intent": "create",
    "entity": "product",
    "data": {"name": "Tablet", "price": 3000},
    "returning": ["id", "name"]
})

---

Operaciones de Conjuntos

UNION

mdbp.query({
    "intent": "union",
    "entity": "customer",
    "union_all": False,
    "union_queries": [
        {"intent": "list", "entity": "customer", "fields": ["name"], "filters": {"city": "Istanbul"}},
        {"intent": "list", "entity": "customer", "fields": ["name"], "filters": {"city": "Ankara"}}
    ]
})

Las intenciones intersect y except también se soportan de la misma manera.

---

Registro de Esquemas

Auto-Descubrimiento (Predeterminado)

mdbp = MDBP(db_url="sqlite:///my.db")
# All tables and columns are automatically registered

Conversión de nombre de tabla a nombre de entidad:

• products -> product
• categories -> category
• order_items -> order_item

Soporte para BigQuery: El controlador SQLAlchemy de BigQuery no puede listar tablas mediante el estándar MetaData.reflect(). MDBP recurre automáticamente a INFORMATION_SCHEMA.TABLES para descubrir tablas y refleja cada una individualmente. No se necesita configuración adicional — solo pasa una URL de BigQuery:

mdbp = MDBP(db_url="bigquery://project-id/dataset")

Registro Manual

Anula el auto-descubrimiento o proporciona nombres personalizados:

from mdbp.core.schema_registry import EntitySchema, FieldSchema

mdbp.register_entity(EntitySchema(
    entity="order",
    table="orders",
    primary_key="id",
    fields={
        "id": FieldSchema(column="id", dtype="integer"),
        "customer_name": FieldSchema(
            column="cust_name",
            dtype="text",
            description="Full name of the customer"
        ),
        "total": FieldSchema(
            column="total_amount",
            dtype="numeric",
            description="Total order amount"
        ),
        "status": FieldSchema(
            column="order_status",
            dtype="text",
            filterable=True,
            sortable=True
        ),
    },
    description="Customer orders"
))

Parámetros de FieldSchema:

Parámetro	Tipo	Predeterminado	Descripción
column	str	-	Nombre de columna física
dtype	str	"text"	Tipo de dato: text, integer, numeric, boolean, datetime
description	str	None	Descripción del campo amigable para LLMs
filterable	bool	True	Puede usarse en filtros
sortable	bool	True	Puede usarse en ordenamiento

Visualización del Esquema

schema = mdbp.describe_schema()

Salida:

{
    "product": {
        "description": "Product catalog",
        "fields": {
            "id": {"type": "integer", "description": null, "filterable": true, "sortable": true},
            "name": {"type": "text", "description": null, "filterable": true, "sortable": true},
            "price": {"type": "numeric", "description": null, "filterable": true, "sortable": true}
        }
    }
}

Esta salida puede incluirse en el prompt del sistema del LLM.

---

Motor de Políticas

El Motor de Políticas proporciona control de acceso basado en roles.

Definición de Políticas

from mdbp.core.policy import Policy

# Analyst: read-only, sensitive fields hidden
mdbp.add_policy(Policy(
    entity="user",
    role="analyst",
    allowed_fields=["id", "name", "email", "created_at"],
    denied_fields=["password_hash", "ssn"],
    max_rows=100,
    allowed_intents=["list", "get", "count"]
))

Parámetros de Política:

Parámetro	Tipo	Predeterminado	Descripción
entity	str	-	Entidad objetivo
role	str	"*"	Nombre del rol ("*" = todos los roles)
allowed_fields	list	None	Campos permitidos (None = todos)
denied_fields	list	[]	Campos denegados (anula permitidos)
max_rows	int	1000	Máximo de filas devueltas
allowed_intents	list	[list,get,count,aggregate]	Operaciones permitidas
row_filter	dict	None	Filtro inyectado automáticamente
masked_fields	dict	{}	Campos a enmascarar en resultados (ver Enmascaramiento de Datos)

Aislamiento de Inquilinos (Tenant)

mdbp.add_policy(Policy(
    entity="order",
    role="customer",
    row_filter={"tenant_id": current_user.tenant_id}
))

Cuando esta política está activa, WHERE tenant_id = :value se agrega automáticamente a todas las consultas. El LLM no puede acceder a los datos de otros inquilinos.

Restricción Global de Intenciones

# Read-only mode
mdbp = MDBP(
    db_url="sqlite:///my.db",
    allowed_intents=["list", "get", "count", "aggregate"]
)

Esto funciona independientemente del motor de políticas. Las intenciones create, update, delete están bloqueadas globalmente.

Consulta con un Rol

result = mdbp.query({
    "intent": "list",
    "entity": "user",
    "fields": ["name", "password_hash"],
    "role": "analyst"
})
# Error: MDBP_POLICY_FIELD_DENIED

---

Enmascaramiento de Datos

El enmascaramiento de datos permite devolver valores enmascarados para campos sensibles en lugar de bloquear la consulta por completo. A diferencia de denied_fields (que rechaza la consulta), masked_fields permite la consulta pero enmascara los valores en la respuesta.

Uso Básico

from mdbp.core.policy import Policy

mdbp.add_policy(Policy(
    entity="customer",
    role="support",
    masked_fields={
        "email": "email",       # d***@example.com
        "phone": "last_n",      # ******4567
    },
    # name, city, etc. are returned unmasked
))

Solo los campos listados en masked_fields se enmascaran. Todos los demás campos se devuelven tal cual.

Estrategias Incorporadas

Estrategia	Descripción	Ejemplo
"partial"	Muestra el primer y último carácter	"doruk" → "d***k"
"redact"	Reemplaza por completo	"doruk" → "***"
"email"	Enmascara la parte local, conserva el dominio	"d@x.com" → "d***@x.com"
"last_n"	Muestra solo los últimos N caracteres (predeterminado 4)	"5551234567" → "******4567"
"first_n"	Muestra solo los primeros N caracteres (predeterminado 4)	"5551234567" → "5551******"
"hash"	Hash SHA-256 (primeros 8 caracteres)	"doruk" → "a1b2c3d4"

Opciones de Estrategia (MaskingRule)

Usa MaskingRule para estrategias que requieren configuración:

from mdbp import MaskingRule

mdbp.add_policy(Policy(
    entity="customer",
    role="support",
    masked_fields={
        "phone": MaskingRule(strategy="last_n", options={"n": 4}),
        "credit_card": MaskingRule(strategy="last_n", options={"n": 4}),
        "email": MaskingRule(strategy="hash", options={"length": 12}),
    },
))

Funciones de Enmascaramiento Personalizadas

Proporciona cualquier invocable para control total:

mdbp.add_policy(Policy(
    entity="customer",
    role="support",
    masked_fields={
        "ssn": lambda v: "***-**-" + str(v)[-4:],
        "email": lambda v: v.split("@")[0][0] + "***@" + v.split("@")[1],
    },
))

Enmascaramiento + Campos Denegados

masked_fields y denied_fields funcionan juntos:

mdbp.add_policy(Policy(
    entity="customer",
    role="support",
    denied_fields=["password_hash"],       # query is rejected if requested
    masked_fields={"email": "email"},      # query succeeds, value is masked
))

Soporte de Archivo de Configuración

{
    "policies": [{
        "entity": "customer",
        "role": "support",
        "masked_fields": {
            "email": "email",
            "phone": {"strategy": "last_n", "options": {"n": 4}}
        }
    }]
}

Notas

• Los valores None/null nunca se enmascaran — permanecen como None
• Los valores numéricos se convierten a string antes de enmascarar
• El enmascaramiento lo aplica la librería después de la ejecución de la consulta, no la IA
• Funciona con todos los tipos de intención que devuelven datos: list, get, aggregate, mutaciones con returning

---

Modo de Prueba (Dry-Run)

Cualquier intención puede incluir "dry_run": true para obtener el SQL compilado y los parámetros sin ejecutar la consulta. La validación de esquema y la aplicación de políticas siguen vigentes.

result = mdbp.query({
    "intent": "list",
    "entity": "product",
    "filters": {"price__gte": 100},
    "fields": ["name", "price"],
    "dry_run": True
})

Salida:

{
    "success": true,
    "intent": "list",
    "entity": "product",
    "dry_run": true,
    "sql": "SELECT products.name, products.price FROM products WHERE products.price >= :price_1",
    "params": {"price_1": 100}
}

Útil para:

• Depuración: Ver el SQL exacto que genera MDBP
• Pruebas: Validar la estructura de la consulta sin tocar la base de datos
• Flujos de aprobación: Revisar consultas antes de la ejecución

---

Servidor MCP

MDBP puede exponerse a Claude, Cursor y otros clientes compatibles con MCP mediante el Protocolo de Contexto de Modelos.

Inicio mediante CLI

# stdio (default) — for Claude Desktop, Cursor, etc.
mdbp-server --db-url "postgresql://user:pass@localhost/mydb"

# SSE — HTTP + Server-Sent Events at /sse
mdbp-server --db-url "postgresql://..." --transport sse --port 8000

# Streamable HTTP — newer MCP HTTP protocol at /mcp
mdbp-server --db-url "postgresql://..." --transport streamable-http --port 8000

# WebSocket — WebSocket at /ws
mdbp-server --db-url "postgresql://..." --transport websocket --port 8000

# With config file
mdbp-server --db-url "sqlite:///my.db" --config config.json --transport sse

Archivo de Configuración

{
    "entities": [
        {
            "entity": "product",
            "table": "products",
            "primary_key": "id",
            "description": "Product catalog",
            "fields": {
                "id": {"column": "id", "dtype": "integer"},
                "name": {"column": "product_name", "dtype": "text", "description": "Product name"},
                "price": {"column": "unit_price", "dtype": "numeric"}
            }
        }
    ],
    "policies": [
        {
            "entity": "product",
            "role": "viewer",
            "allowed_intents": ["list", "get", "count"],
            "max_rows": 100
        }
    ]
}

Integración con Claude Desktop

claude_desktop_config.json:

{
    "mcpServers": {
        "mdbp": {
            "command": "python",
            "args": ["-u", "path/to/server.py"],
            "env": {
                "PYTHONPATH": "path/to/mdbp/project"
            }
        }
    }
}

Uso Programático

Todos los transportes están disponibles como funciones de una línea:

from mdbp import MDBP
from mdbp.transport.server import run_sse, run_streamable_http, run_websocket, run_stdio

mdbp = MDBP(db_url="postgresql://user:pass@localhost/mydb")

run_sse(mdbp, host="0.0.0.0", port=8000)                # SSE at /sse
run_streamable_http(mdbp, host="0.0.0.0", port=8000)     # Streamable HTTP at /mcp
run_websocket(mdbp, host="0.0.0.0", port=8000)           # WebSocket at /ws
run_stdio(mdbp)                                           # stdin/stdout

Aplicaciones ASGI (para middleware personalizado o montaje):

from mdbp.transport.server import sse_app, streamable_http_app, websocket_app

app = sse_app(mdbp)                # Starlette ASGI app — /sse endpoint
app = streamable_http_app(mdbp)    # Starlette ASGI app — /mcp endpoint
app = websocket_app(mdbp)          # Starlette ASGI app — /ws endpoint

Bajo nivel (control total):

from mdbp.transport.server import create_server

server = create_server(mdbp)  # Returns mcp.server.Server — wire any transport yourself

Herramientas MCP expuestas

Herramienta	Descripción
mdbp_query	Ejecutar una consulta de base de datos basada en intenciones
mdbp_describe_schema	Listar entidades y campos disponibles

---

Manejo de errores

MDBP captura todos los errores y devuelve JSON estructurado. Nunca lanza excepciones desde query().

Estructura de error

{
    "success": false,
    "intent": "list",
    "entity": "product",
    "error": {
        "code": "MDBP_SCHEMA_FIELD_NOT_FOUND",
        "message": "Field 'colour' not found on entity 'product'.",
        "details": {
            "entity": "product",
            "field": "colour",
            "available_fields": ["id", "name", "price", "color", "category_id"]
        }
    }
}

Códigos de error

Errores de esquema (MDBP_SCHEMA_*)

Código	Significado	Detalles
MDBP_SCHEMA_ENTITY_NOT_FOUND	La entidad no existe en el registro	lista available_entities
MDBP_SCHEMA_FIELD_NOT_FOUND	El campo no existe en la entidad	lista available_fields
MDBP_SCHEMA_ENTITY_REF_NOT_FOUND	Entidad JOIN referenciada no encontrada	entity_reference, field

Errores de política (MDBP_POLICY_*)

Código	Significado	Detalles
MDBP_POLICY_INTENT_NOT_ALLOWED	Tipo de intención no permitido para el rol	intent_type, entity, role
MDBP_POLICY_FIELD_DENIED	El campo está en la lista denied_fields	entity, denied_fields
MDBP_POLICY_FIELD_NOT_ALLOWED	El campo no está en la lista allowed_fields	entity, allowed_fields

Errores de intención (MDBP_INTENT_*)

Código	Significado	Detalles
MDBP_INTENT_TYPE_NOT_ALLOWED	Tipo de intención bloqueado globalmente	intent_type, allowed_intents
MDBP_INTENT_VALIDATION_ERROR	Estructura de intención inválida (Pydantic)	lista de errores

Errores de consulta (MDBP_QUERY_*)

Código	Significado	Detalles
MDBP_QUERY_PLAN_ERROR	Falló la planificación de la consulta	-
MDBP_QUERY_MISSING_FIELD	Falta un campo requerido	intent_type, required_field
MDBP_QUERY_UNKNOWN_FILTER_OP	Operador de filtro desconocido	op, supported_ops
MDBP_QUERY_UNION_REQUIRES_SUBQUERIES	UNION necesita 2+ subconsultas	-

Errores de conexión (MDBP_CONN_*)

Código	Significado	Detalles
MDBP_CONN_FAILED	Falló la conexión a la base de datos	-
MDBP_CONN_EXECUTION_ERROR	Falló la ejecución de la consulta	original_error
MDBP_NOT_FOUND	La consulta GET no devolvió resultados	entity, id

Errores de configuración

Código	Significado	Detalles
MDBP_CONFIG_FILE_NOT_FOUND	El archivo de configuración no existe	path

Manejo de errores en código

from mdbp import MDBP

mdbp = MDBP(db_url="sqlite:///my.db")
result = mdbp.query({"intent": "list", "entity": "product"})

if not result["success"]:
    code = result["error"]["code"]
    if code == "MDBP_SCHEMA_ENTITY_NOT_FOUND":
        entities = result["error"]["details"]["available_entities"]
        print(f"Available entities: {entities}")

---

Referencia de la API

Clase MDBP

class MDBP:
    def __init__(
        self,
        db_url: str,
        auto_discover: bool = True,
        allowed_intents: list[str] | None = None,
    ) -> None

    def register_entity(schema: EntitySchema) -> None
    def add_policy(policy: Policy) -> None
    def query(raw_intent: dict | Intent) -> dict
    def describe_schema() -> dict
    def dispose() -> None

Método	Descripción
register_entity()	Registrar un esquema de entidad personalizado (anula el auto-descubrimiento)
add_policy()	Agregar una política de control de acceso
query()	Ejecutar el pipeline completo de MDBP. Acepta dict o Intent. Devuelve una respuesta estructurada.
describe_schema()	Devolver una descripción de esquema amigable para LLM
dispose()	Liberar todas las conexiones de base de datos

EntitySchema

class EntitySchema(BaseModel):
    entity: str                           # Logical entity name
    table: str                            # Physical table name
    primary_key: str = "id"               # Primary key column
    fields: dict[str, FieldSchema]        # Field definitions
    relations: dict[str, RelationSchema] = {}
    description: str | None = None

FieldSchema

class FieldSchema(BaseModel):
    column: str              # Physical column name
    dtype: str = "text"      # text, integer, numeric, boolean, datetime
    description: str | None = None
    filterable: bool = True
    sortable: bool = True

RelationSchema

class RelationSchema(BaseModel):
    target_entity: str                  # Related entity name
    join_column: str                    # Column on this entity's table
    target_column: str                  # Column on target entity's table
    relation_type: str = "many_to_one"  # one_to_one, many_to_one, one_to_many

Policy

class Policy(BaseModel):
    entity: str
    role: str = "*"
    allowed_fields: list[str] | None = None
    denied_fields: list[str] = []
    max_rows: int = 1000
    allowed_intents: list[IntentType] = [LIST, GET, COUNT, AGGREGATE]
    row_filter: dict | None = None
    masked_fields: dict[str, str | MaskingRule | Callable] = {}

IntentType Enum

class IntentType(str, Enum):
    LIST = "list"
    GET = "get"
    COUNT = "count"
    AGGREGATE = "aggregate"
    CREATE = "create"
    BATCH_CREATE = "batch_create"
    UPSERT = "upsert"
    UPDATE = "update"
    DELETE = "delete"
    UNION = "union"
    INTERSECT = "intersect"
    EXCEPT = "except"

---

Seguridad

Protección contra alucinaciones

Los LLM pueden generar nombres de tablas o columnas inexistentes. El registro de esquemas los detecta:

LLM: query "userz" table
MDBP: MDBP_SCHEMA_ENTITY_NOT_FOUND + list of available entities
LLM: self-corrects -> query "user" table

Prevención de inyección SQL

Todas las consultas se parametrizan mediante SQLAlchemy. Nunca se construyen cadenas SQL sin procesar.

Control de acceso

• denied_fields: Los campos sensibles (password_hash, ssn) nunca pueden ser devueltos
• allowed_fields: Solo los campos en la lista blanca son accesibles
• masked_fields: Los campos sensibles se devuelven con valores enmascarados (email, phone, etc.)
• allowed_intents: Las operaciones de escritura pueden bloquearse globalmente o por rol
• max_rows: Limita los resultados de consultas grandes por rol
• row_filter: Aislamiento automático de inquilinos mediante condiciones WHERE inyectadas

---

Pruébalo

El proyecto examples/ecommerce-mdbp-server es un ejemplo completo y funcional que puedes ejecutar localmente:

cd examples/ecommerce-mdbp-server
pip install mdbp
python setup_db.py    # Create e-commerce database with sample data
python server.py      # Start MDBP server on :8000

Este ejemplo demuestra:

• Auto-descubrimiento (sin registro manual de esquemas)
• Control de acceso basado en roles (customer, support, admin)
• Aislamiento de inquilinos mediante row_filter
• Protección de PII mediante denied_fields
• Los 4 modos de transporte (stdio, sse, streamable-http, websocket)

Consulta el README del ejemplo para obtener detalles completos y consultas de muestra.

---

_{MDBP — Model Database Protocol}