Model Database Protocol

Intent-based data access protocol for AI systems.

MDBP enables secure database access for LLMs. Instead of generating raw SQL, LLMs produce structuredintentobjects. MDBP validates these intents against a schema registry, enforces access policies, builds parameterized queries via SQLAlchemy, and returns LLM-friendly responses.

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

- Installation
- Quick Start
- Zero-Config Database Connection
- Core Concepts
- Intent Types
- Filtering
- JOIN Operations
- Aggregation
- Computed Fields
- CTE (Common Table Expressions)
- Write Operations
- Set Operations
- Schema Registry
- Policy Engine
- Data Masking
- Dry-Run Mode
- MCP Server
- Error Handling
- API Reference
- Security

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

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

{ "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"] } } }

When an LLM hallucinates a table name, MDBP catches it and returns the list of available entities. The LLM can self-correct using this feedback.

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()

MDBP supports any SQLAlchemy-compatible database with zero code. Just install the driver and connect:

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

All tables, columns, and types are auto-discovered — no schema definition or server code needed.

# 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

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

An intent is a structured JSON object that describes a database operation. Every intent contains these core fields:

Everymdbp.query()call passes through these stages:

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

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

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

Returns a single record by primary key. ReturnsMDBP_NOT_FOUNDerror if no record exists.

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

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

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

Supported operations:sum,avg,min,max,count

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

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

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

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

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

Append a suffix to the field name in thefiltersdict to specify the operator:

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 } })

Use thewherefield for nested AND/OR/NOT logic:

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} ] } ] } })

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

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

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

Use$queryin filter values for subqueries:

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

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

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

- on:{local_field: foreign_field}format
- Dot notation infields:"customer.name"resolves to the joined table's column
- type:inner,left,right,full

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"}} ] })

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

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

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

# 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"], []] })

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" } }] })

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"}] } }] })

Supported window functions:rank,dense_rank,row_number,ntile,lag,lead,first_value,last_value,sum,avg,min,max,count

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"]} } ] })

Supported scalar functions:coalesce,upper,lower,cast,concat,trim,length,abs,round,substring,extract,now,current_date,replace

Note:extractrequires the first argument as{"literal": "part"}where part isyear,month,day,hour,minute, orsecond.

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"} } })

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

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=...

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

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

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"}} ] })

intersectandexceptintents are also supported in the same way.

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

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

BigQuery support:BigQuery's SQLAlchemy driver can't list tables via standardMetaData.reflect(). MDBP automatically falls back toINFORMATION_SCHEMA.TABLESto discover tables and reflects each one individually. No extra configuration needed — just pass a BigQuery URL:

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

Override auto-discovery or provide custom names:

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" ))

{ "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} } } }

This output can be included in an LLM system prompt.

The Policy Engine provides role-based access control.

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"] ))

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

When this policy is active,WHERE tenant_id = :valueis automatically appended to all queries. The LLM cannot access other tenants' data.

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

This works independently from the policy engine.create,update,deleteintents are globally blocked.

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

Data masking lets you return masked values for sensitive fields instead of blocking the query entirely. Unlikedenied_fields(which rejects the query),masked_fieldsallows the query but masks the values in the response.