mdbp 0.3.4.7

Model Database Protocol — intent-based, secure data access for AI systems

MDBP - 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 structured intent objects. 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

---

Table of Contents

• 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

---

Installation

pip install mdbp

For development:

pip install mdbp[dev]

Requirements:

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

Supported Databases: Any SQLAlchemy-supported backend: PostgreSQL, MySQL, SQLite, MSSQL, Oracle, BigQuery, etc.

---

Quick Start

Up and Running in 3 Lines

from mdbp import MDBP

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

When MDBP(db_url=...) is called, all tables and columns are automatically discovered from the database. No manual registration required.

Example Output

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

Error Response

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

Real-World Example (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()

---

Zero-Config Database Connection

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.

Examples:

# 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 / MCP Clients:

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

---

Core Concepts

What is an Intent?

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

Field	Type	Required	Description
intent	string	Yes	Operation type: list, get, count, aggregate, create, update, delete
entity	string	Yes	Target table/entity name
filters	object	No	Filter conditions
fields	array	No	Fields to return (empty = all)
sort	array	No	Ordering
limit	integer	No	Result limit
offset	integer	No	Pagination offset

Pipeline

Every mdbp.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

---

Intent Types

list - List Records

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 - Get Single Record

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

Returns a single record by primary key. Returns MDBP_NOT_FOUND error if no record exists.

count - Count Records

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

Output:

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

aggregate - Aggregate

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

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

Multiple aggregations:

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

create - Create Record

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

update - Update Record

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

Bulk update with filters:

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

delete - Delete Record

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

---

Filtering

Simple Filters (Operator Suffix)

Append a suffix to the field name in the filters dict 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
    }
})

All Operators:

Suffix	SQL Equivalent	Example
(none)	=	{"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}

Complex Filters (where)

Use the where field 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}
                ]
            }
        ]
    }
})

SQL equivalent:

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

NOT example:

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

EXISTS example:

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

Subquery Filters

Use $query in filter values for subqueries:

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

SQL equivalent:

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

---

JOIN Operations

Basic JOIN

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 in fields: "customer.name" resolves to the joined table's column
• type: inner, left, right, full

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

Self-JOIN (Alias)

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

---

Aggregation

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

Advanced GROUP BY Modes

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

---

Computed Fields

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

Window Functions

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

Scalar Functions

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: extract requires the first argument as {"literal": "part"} where part is year, month, day, hour, minute, or second.

---

CTE (Common Table Expressions)

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

---

Write Operations

batch_create - Bulk Insert

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

upsert - Insert or Update

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

---

Set Operations

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

intersect and except intents are also supported in the same way.

---

Schema Registry

Auto-Discovery (Default)

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

Table name to entity name conversion:

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

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

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

Manual Registration

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

FieldSchema Parameters:

Parameter	Type	Default	Description
column	str	-	Physical column name
dtype	str	"text"	Data type: text, integer, numeric, boolean, datetime
description	str	None	LLM-friendly field description
filterable	bool	True	Can be used in filters
sortable	bool	True	Can be used in sort

Viewing the Schema

schema = mdbp.describe_schema()

Output:

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

---

Policy Engine

The Policy Engine provides role-based access control.

Defining Policies

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

Policy Parameters:

Parameter	Type	Default	Description
entity	str	-	Target entity
role	str	"*"	Role name ("*" = all roles)
allowed_fields	list	None	Allowed fields (None = all)
denied_fields	list	[]	Denied fields (overrides allowed)
max_rows	int	1000	Maximum rows returned
allowed_intents	list	[list,get,count,aggregate]	Allowed operations
row_filter	dict	None	Automatically injected filter
masked_fields	dict	{}	Fields to mask in results (see Data Masking)

Tenant Isolation

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

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

Global Intent Restriction

# 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, delete intents are globally blocked.

Querying with a Role

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

---

Data Masking

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

Basic Usage

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

Only the fields listed in masked_fields are masked. All other fields are returned as-is.

Built-in Strategies

Strategy	Description	Example
"partial"	Show first and last character	"doruk" → "d***k"
"redact"	Replace entirely	"doruk" → "***"
"email"	Mask local part, keep domain	"[e-mail]" → "d***@x.com"
"last_n"	Show only last N characters (default 4)	"5551234567" → "******4567"
"first_n"	Show only first N characters (default 4)	"5551234567" → "5551******"
"hash"	SHA-256 hash (first 8 chars)	"doruk" → "a1b2c3d4"

Strategy Options (MaskingRule)

Use MaskingRule for strategies that need configuration:

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

Custom Masking Functions

Provide any callable for full control:

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

Masking + Denied Fields

masked_fields and denied_fields work together:

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

Config File Support

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

Notes

• None/null values are never masked — they stay None
• Numeric values are converted to string before masking
• Masking is applied by the library after query execution, not by the AI
• Works with all intent types that return data: list, get, aggregate, mutations with returning

---

Dry-Run Mode

Any intent can include "dry_run": true to get the compiled SQL and parameters without executing the query. Schema validation and policy enforcement still apply.

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

Output:

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

Useful for:

• Debugging: See the exact SQL that MDBP generates
• Testing: Validate query structure without hitting the database
• Approval workflows: Review queries before execution

---

MCP Server

MDBP can be exposed to Claude, Cursor, and other MCP-compatible clients via the Model Context Protocol.

Starting via 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

Config File

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

Claude Desktop Integration

claude_desktop_config.json:

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

Programmatic Usage

All transports are available as one-liner functions:

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

ASGI apps (for custom middleware or mounting):

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

Low-level (full control):

from mdbp.transport.server import create_server

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

Exposed MCP Tools

Tool	Description
mdbp_query	Execute an intent-based database query
mdbp_describe_schema	List available entities and fields

---

Error Handling

MDBP catches all errors and returns structured JSON. It never raises exceptions from query().

Error Structure

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

Error Codes

Schema Errors (MDBP_SCHEMA_*)

Code	Meaning	Details
MDBP_SCHEMA_ENTITY_NOT_FOUND	Entity does not exist in registry	available_entities list
MDBP_SCHEMA_FIELD_NOT_FOUND	Field does not exist on entity	available_fields list
MDBP_SCHEMA_ENTITY_REF_NOT_FOUND	Referenced JOIN entity not found	entity_reference, field

Policy Errors (MDBP_POLICY_*)

Code	Meaning	Details
MDBP_POLICY_INTENT_NOT_ALLOWED	Intent type not allowed for role	intent_type, entity, role
MDBP_POLICY_FIELD_DENIED	Field is in denied_fields list	entity, denied_fields
MDBP_POLICY_FIELD_NOT_ALLOWED	Field is not in allowed_fields list	entity, allowed_fields

Intent Errors (MDBP_INTENT_*)

Code	Meaning	Details
MDBP_INTENT_TYPE_NOT_ALLOWED	Intent type globally blocked	intent_type, allowed_intents
MDBP_INTENT_VALIDATION_ERROR	Invalid intent structure (Pydantic)	errors list

Query Errors (MDBP_QUERY_*)

Code	Meaning	Details
MDBP_QUERY_PLAN_ERROR	Query planning failed	-
MDBP_QUERY_MISSING_FIELD	Required field missing	intent_type, required_field
MDBP_QUERY_UNKNOWN_FILTER_OP	Unknown filter operator	op, supported_ops
MDBP_QUERY_UNION_REQUIRES_SUBQUERIES	UNION needs 2+ sub-queries	-

Connection Errors (MDBP_CONN_*)

Code	Meaning	Details
MDBP_CONN_FAILED	Database connection failed	-
MDBP_CONN_EXECUTION_ERROR	Query execution failed	original_error
MDBP_NOT_FOUND	GET query returned no results	entity, id

Config Errors

Code	Meaning	Details
MDBP_CONFIG_FILE_NOT_FOUND	Config file does not exist	path

Handling Errors in Code

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

---

API Reference

MDBP Class

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

Method	Description
register_entity()	Register a custom entity schema (overrides auto-discovery)
add_policy()	Add an access control policy
query()	Execute the full MDBP pipeline. Accepts dict or Intent. Returns structured response.
describe_schema()	Return LLM-friendly schema description
dispose()	Release all database connections

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"

---

Security

Hallucination Protection

LLMs can generate non-existent table or column names. The schema registry catches these:

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

SQL Injection Prevention

All queries are parameterized via SQLAlchemy. Raw SQL strings are never constructed.

Access Control

• denied_fields: Sensitive fields (password_hash, ssn) can never be returned
• allowed_fields: Only whitelisted fields are accessible
• masked_fields: Sensitive fields are returned with masked values (email, phone, etc.)
• allowed_intents: Write operations can be blocked globally or per role
• max_rows: Limits large query results per role
• row_filter: Automatic tenant isolation via injected WHERE conditions

---

Try It Out

The examples/ecommerce-mdbp-server project is a complete working example you can run locally:

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

This example demonstrates:

• Auto-discovery (no manual schema registration)
• Role-based access control (customer, support, admin)
• Tenant isolation via row_filter
• PII protection via denied_fields
• All 4 transport modes (stdio, sse, streamable-http, websocket)

See the example README for full details and sample queries.

---

_{MDBP — Model Database Protocol}