Model Database Protocol

Intent-based, secure database access protocol for AI systems — LLMs send structured intents instead of raw SQL.

GitHub

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

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

Core Concepts

What is an Intent?

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

FieldTypeRequiredDescription
intentstringYesOperation type: list, get, count, aggregate, create, update, delete
entitystringYesTarget table/entity name
filtersobjectNoFilter conditions
fieldsarrayNoFields to return (empty = all)
sortarrayNoOrdering
limitintegerNoResult limit
offsetintegerNoPagination 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:

SuffixSQL EquivalentExample
(none)={"city": "Istanbul"}
__gt>{"price__gt": 100}
__gte>={"price__gte": 100}
__lt<{"price__lt": 500}
__lte<={"price__lte": 500}
__ne!={"status__ne": "deleted"}
__likeLIKE{"name__like": "%phone%"}
__ilikeILIKE{"name__ilike": "%Phone%"}
__not_likeNOT LIKE{"name__not_like": "%test%"}
__inIN (...){"id__in": [1, 2, 3]}
__not_inNOT IN{"id__not_in": [4, 5]}
__betweenBETWEEN{"price__between": [100, 500]}
__nullIS NULL{"email__null": true}
__not_nullIS 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

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:

ParameterTypeDefaultDescription
columnstr-Physical column name
dtypestr"text"Data type: text, integer, numeric, boolean, datetime
descriptionstrNoneLLM-friendly field description
filterableboolTrueCan be used in filters
sortableboolTrueCan 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:

ParameterTypeDefaultDescription
entitystr-Target entity
rolestr"*"Role name ("*" = all roles)
allowed_fieldslistNoneAllowed fields (None = all)
denied_fieldslist[]Denied fields (overrides allowed)
max_rowsint1000Maximum rows returned
allowed_intentslist[list,get,count,aggregate]Allowed operations
row_filterdictNoneAutomatically injected filter
masked_fieldsdict{}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

StrategyDescriptionExample
"partial"Show first and last character"doruk" → "d***k"
"redact"Replace entirely"doruk" → "***"
"email"Mask local part, keep domain"[email protected]" → "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

ToolDescription
mdbp_queryExecute an intent-based database query
mdbp_describe_schemaList 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_*)

CodeMeaningDetails
MDBP_SCHEMA_ENTITY_NOT_FOUNDEntity does not exist in registryavailable_entities list
MDBP_SCHEMA_FIELD_NOT_FOUNDField does not exist on entityavailable_fields list
MDBP_SCHEMA_ENTITY_REF_NOT_FOUNDReferenced JOIN entity not foundentity_reference, field

Policy Errors (MDBP_POLICY_*)

CodeMeaningDetails
MDBP_POLICY_INTENT_NOT_ALLOWEDIntent type not allowed for roleintent_type, entity, role
MDBP_POLICY_FIELD_DENIEDField is in denied_fields listentity, denied_fields
MDBP_POLICY_FIELD_NOT_ALLOWEDField is not in allowed_fields listentity, allowed_fields

Intent Errors (MDBP_INTENT_*)

CodeMeaningDetails
MDBP_INTENT_TYPE_NOT_ALLOWEDIntent type globally blockedintent_type, allowed_intents
MDBP_INTENT_VALIDATION_ERRORInvalid intent structure (Pydantic)errors list

Query Errors (MDBP_QUERY_*)

CodeMeaningDetails
MDBP_QUERY_PLAN_ERRORQuery planning failed-
MDBP_QUERY_MISSING_FIELDRequired field missingintent_type, required_field
MDBP_QUERY_UNKNOWN_FILTER_OPUnknown filter operatorop, supported_ops
MDBP_QUERY_UNION_REQUIRES_SUBQUERIESUNION needs 2+ sub-queries-

Connection Errors (MDBP_CONN_*)

CodeMeaningDetails
MDBP_CONN_FAILEDDatabase connection failed-
MDBP_CONN_EXECUTION_ERRORQuery execution failedoriginal_error
MDBP_NOT_FOUNDGET query returned no resultsentity, id

Config Errors

CodeMeaningDetails
MDBP_CONFIG_FILE_NOT_FOUNDConfig file does not existpath

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
MethodDescription
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 Logo
MDBP — Model Database Protocol

関連サーバー

NotebookLM Webインポーター

ワンクリックでWebページとYouTube動画をNotebookLMにインポート。200,000人以上のユーザーが利用中。

Chrome拡張機能をインストール