Servidor MCP Model Database Protocol

MDBP - Protocolo de Banco de Dados Modelo

Protocolo de acesso a dados baseado em intenção para sistemas de IA.

O MDBP permite acesso seguro a banco de dados para LLMs. Em vez de gerar SQL bruto, LLMs produzem objetos estruturados de intenção. O MDBP valida essas intenções contra um registro de esquema, aplica políticas de acesso, constrói consultas parametrizadas via SQLAlchemy e retorna respostas amigáveis para LLMs.

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

Sumário

Instalação
Início Rápido
Conexão de Banco de Dados Zero-Config
Conceitos Principais
Tipos de Intenção
Filtragem
Operações JOIN
Agregação
Campos Calculados
CTE (Common Table Expressions)
Operações de Escrita
Operações de Conjunto
Registro de Esquema
Mecanismo de Políticas
Mascaramento de Dados
Modo Dry-Run
Servidor MCP
Tratamento de Erros
Referência da API
Segurança

Instalação

pip install mdbp

Para desenvolvimento:

pip install mdbp[dev]

Requisitos:

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

Bancos de Dados Suportados: Qualquer backend suportado pelo SQLAlchemy: PostgreSQL, MySQL, SQLite, MSSQL, Oracle, BigQuery, etc.

Início Rápido

Funcionando em 3 Linhas

from mdbp import MDBP

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

Quando MDBP(db_url=...) é chamado, todas as tabelas e colunas são automaticamente descobertas do banco de dados. Nenhum registro manual é necessário.

Exemplo de Saída

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

Resposta de Erro

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

Quando um LLM alucina um nome de tabela, o MDBP captura isso e retorna a lista de entidades disponíveis. O LLM pode se autocorrigir usando esse feedback.

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

Conexão de Banco de Dados Zero-Config

O MDBP suporta qualquer banco de dados compatível com SQLAlchemy com zero código. Basta instalar o driver e conectar:

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

Todas as tabelas, colunas e tipos são descobertos automaticamente — nenhuma definição de esquema ou código de servidor é necessária.

Exemplos:

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

Conceitos Principais

O que é uma Intenção?

Uma intenção é um objeto JSON estruturado que descreve uma operação de banco de dados. Toda intenção contém estes campos principais:

Campo	Tipo	Obrigatório	Descrição
`intent`	string	Sim	Tipo de operação: list, get, count, aggregate, create, update, delete
`entity`	string	Sim	Nome da tabela/entidade alvo
`filters`	objeto	Não	Condições de filtro
`fields`	array	Não	Campos a retornar (vazio = todos)
`sort`	array	Não	Ordenação
`limit`	inteiro	Não	Limite de resultados
`offset`	inteiro	Não	Deslocamento de paginação

Pipeline

Toda chamada mdbp.query() passa 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 Intenção

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 - Obter Registro Único

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

Retorna um único registro pela chave primária. Retorna erro MDBP_NOT_FOUND se nenhum registro existir.

count - Contar Registros

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

Saída:

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

aggregate - Agregar

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

Operações suportadas: sum, avg, min, max, count

Múltiplas agregações:

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

create - Criar Registro

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

update - Atualizar Registro

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

Atualização em massa com filtros:

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

delete - Excluir Registro

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

Filtragem

Filtros Simples (Sufixo de Operador)

Anexe um sufixo ao nome do campo no dicionário filters para especificar o 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 os Operadores:

Sufixo	Equivalente SQL	Exemplo
(nenhum)	`=`	`{"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 Complexos (where)

Use o campo where para lógica aninhada 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)

Exemplo NOT:

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

Exemplo EXISTS:

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

Filtros de Subconsulta

Use $query em 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')

Operações 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}
Notação de ponto em fields: "customer.name" resolve para a coluna da tabela unida
type: inner, left, right, full

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

Agregação

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

Funções de Janela

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

Funções de janela suportadas: rank, dense_rank, row_number, ntile, lag, lead, first_value, last_value, sum, avg, min, max, count

Funções 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"]}
        }
    ]
})

Funções escalares suportadas: coalesce, upper, lower, cast, concat, trim, length, abs, round, substring, extract, now, current_date, replace

Nota: extract requer o primeiro argumento como {"literal": "part"} onde a parte é year, month, day, hour, minute, ou 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"}
    }
})

Operações de Escrita

batch_create - Inserção em Massa

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

upsert - Inserir ou Atualizar

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

Operações de Conjunto

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

Intenções intersect e except também são suportadas da mesma forma.

Registro de Esquema

Auto-Descoberta (Padrão)

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

Conversão de nome de tabela para nome de entidade:

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

Suporte BigQuery: O driver SQLAlchemy do BigQuery não pode listar tabelas via MetaData.reflect() padrão. O MDBP automaticamente recorre a INFORMATION_SCHEMA.TABLES para descobrir tabelas e reflete cada uma individualmente. Nenhuma configuração extra é necessária — basta passar uma URL do BigQuery:

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

Registro Manual

Substitua a auto-descoberta ou forneça nomes 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 do FieldSchema:

Parâmetro	Tipo	Padrão	Descrição
`column`	str	-	Nome da coluna física
`dtype`	str	"text"	Tipo de dados: text, integer, numeric, boolean, datetime
`description`	str	None	Descrição do campo amigável para LLM
`filterable`	bool	True	Pode ser usado em filtros
`sortable`	bool	True	Pode ser usado em ordenação

Visualizando o Esquema

schema = mdbp.describe_schema()

Saída:

{
    "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 saída pode ser incluída no prompt de sistema de um LLM.

Mecanismo de Políticas

O Mecanismo de Políticas fornece controle de acesso baseado em papéis.

Definindo 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	Padrão	Descrição
`entity`	str	-	Entidade alvo
`role`	str	"*"	Nome do papel ("*" = todos os papéis)
`allowed_fields`	list	None	Campos permitidos (None = todos)
`denied_fields`	list	[]	Campos negados (sobrepõe permitidos)
`max_rows`	int	1000	Máximo de linhas retornadas
`allowed_intents`	list	[list,get,count,aggregate]	Operações permitidas
`row_filter`	dict	None	Filtro injetado automaticamente
`masked_fields`	dict	{}	Campos a mascarar nos resultados (veja Mascaramento de Dados)

Isolamento de Locatário

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

Quando esta política está ativa, WHERE tenant_id = :value é automaticamente anexado a todas as consultas. O LLM não pode acessar dados de outros locatários.

Restrição Global de Intenção

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

Isso funciona independentemente do mecanismo de políticas. Intenções create, update, delete são bloqueadas globalmente.

Consultando com um Papel

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

Mascaramento de Dados

O mascaramento de dados permite retornar valores mascarados para campos sensíveis em vez de bloquear a consulta inteira. Diferente de denied_fields (que rejeita a consulta), masked_fields permite a consulta, mas mascara os valores na resposta.

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

Apenas os campos listados em masked_fields são mascarados. Todos os outros campos são retornados como estão.

Estratégias Integradas

Estratégia	Descrição	Exemplo
`"partial"`	Mostra o primeiro e o último caractere	`"doruk"` → `"d***k"`
`"redact"`	Substitui completamente	`"doruk"` → `"***"`
`"email"`	Mascara a parte local, mantém o domínio	`"d@x.com"` → `"d***@x.com"`
`"last_n"`	Mostra apenas os últimos N caracteres (padrão 4)	`"5551234567"` → `"******4567"`
`"first_n"`	Mostra apenas os primeiros N caracteres (padrão 4)	`"5551234567"` → `"5551******"`
`"hash"`	Hash SHA-256 (primeiros 8 caracteres)	`"doruk"` → `"a1b2c3d4"`

Opções de Estratégia (MaskingRule)

Use MaskingRule para estratégias que precisam de configuração:

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

Funções de Mascaramento Personalizadas

Forneça qualquer chamável para controle 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],
    },
))

Mascaramento + Campos Negados

masked_fields e denied_fields funcionam 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
))

Suporte a Arquivo de Configuração

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

Notas

Valores None/null nunca são mascarados — eles permanecem None
Valores numéricos são convertidos para string antes do mascaramento
O mascaramento é aplicado pela biblioteca após a execução da consulta, não pela IA
Funciona com todos os tipos de intenção que retornam dados: list, get, aggregate, mutações com returning

Modo Dry-Run

Qualquer intenção pode incluir "dry_run": true para obter o SQL compilado e os parâmetros sem executar a consulta. A validação de esquema e a aplicação de políticas ainda se aplicam.

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

Saída:

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

Depuração: Veja o SQL exato que o MDBP gera
Testes: Valide a estrutura da consulta sem acessar o banco de dados
Fluxos de aprovação: Revise consultas antes da execução

Servidor MCP

O MDBP pode ser exposto ao Claude, Cursor e outros clientes compatíveis com MCP via Model Context Protocol.

Iniciando 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

Arquivo de Configuração

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

Integração com 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 os transportes estão disponíveis como funções de uma linha:

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

Aplicativos ASGI (para middleware personalizado ou montagem):

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

Baixo nível (controle total):

from mdbp.transport.server import create_server

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

Ferramentas MCP Expostas

Ferramenta	Descrição
`mdbp_query`	Executa uma consulta de banco de dados baseada em intenção
`mdbp_describe_schema`	Lista entidades e campos disponíveis

Tratamento de Erros

O MDBP captura todos os erros e retorna JSON estruturado. Ele nunca levanta exceções de query().

Estrutura de Erro

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

Erros de Esquema (`MDBP_SCHEMA_*`)

Código	Significado	Detalhes
`MDBP_SCHEMA_ENTITY_NOT_FOUND`	Entidade não existe no registro	lista de entidades disponíveis
`MDBP_SCHEMA_FIELD_NOT_FOUND`	Campo não existe na entidade	lista de campos disponíveis
`MDBP_SCHEMA_ENTITY_REF_NOT_FOUND`	Entidade JOIN referenciada não encontrada	entity_reference, field

Erros de Política (`MDBP_POLICY_*`)

Código	Significado	Detalhes
`MDBP_POLICY_INTENT_NOT_ALLOWED`	Tipo de intenção não permitido para o papel	intent_type, entity, role
`MDBP_POLICY_FIELD_DENIED`	Campo está na lista denied_fields	entity, denied_fields
`MDBP_POLICY_FIELD_NOT_ALLOWED`	Campo não está na lista allowed_fields	entity, allowed_fields

Erros de Intenção (`MDBP_INTENT_*`)

Código	Significado	Detalhes
`MDBP_INTENT_TYPE_NOT_ALLOWED`	Tipo de intenção bloqueado globalmente	intent_type, allowed_intents
`MDBP_INTENT_VALIDATION_ERROR`	Estrutura de intenção inválida (Pydantic)	lista de erros

Erros de Consulta (`MDBP_QUERY_*`)

Código	Significado	Detalhes
`MDBP_QUERY_PLAN_ERROR`	Falha no planejamento da consulta	-
`MDBP_QUERY_MISSING_FIELD`	Campo obrigatório ausente	intent_type, required_field
`MDBP_QUERY_UNKNOWN_FILTER_OP`	Operador de filtro desconhecido	op, supported_ops
`MDBP_QUERY_UNION_REQUIRES_SUBQUERIES`	UNION precisa de 2+ subconsultas	-

Erros de Conexão (`MDBP_CONN_*`)

Código	Significado	Detalhes
`MDBP_CONN_FAILED`	Falha na conexão com o banco de dados	-
`MDBP_CONN_EXECUTION_ERROR`	Falha na execução da consulta	original_error
`MDBP_NOT_FOUND`	Consulta GET não retornou resultados	entity, id

Erros de Configuração

Código	Significado	Detalhes
`MDBP_CONFIG_FILE_NOT_FOUND`	Arquivo de configuração não existe	path

Tratamento de Erros no 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}")

Referência da API

Classe 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	Descrição
`register_entity()`	Registra um esquema de entidade personalizado (substitui a descoberta automática)
`add_policy()`	Adiciona uma política de controle de acesso
`query()`	Executa o pipeline completo do MDBP. Aceita dict ou Intent. Retorna resposta estruturada.
`describe_schema()`	Retorna descrição de esquema amigável para LLM
`dispose()`	Libera todas as conexões com o banco de dados

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

Enum IntentType

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"

Segurança

Proteção contra Alucinação

LLMs podem gerar nomes de tabelas ou colunas inexistentes. O registro de esquema captura esses casos:

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

Prevenção de Injeção de SQL

Todas as consultas são parametrizadas via SQLAlchemy. Strings SQL brutas nunca são construídas.

Controle de Acesso

denied_fields: Campos sensíveis (password_hash, ssn) nunca podem ser retornados
allowed_fields: Apenas campos na lista de permissões são acessíveis
masked_fields: Campos sensíveis são retornados com valores mascarados (email, telefone, etc.)
allowed_intents: Operações de escrita podem ser bloqueadas globalmente ou por papel
max_rows: Limita resultados de consultas grandes por papel
row_filter: Isolamento automático de tenant via condições WHERE injetadas

Experimente

O projeto examples/ecommerce-mdbp-server é um exemplo completo e funcional que você pode executar 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 exemplo demonstra:

Descoberta automática (sem registro manual de esquema)
Controle de acesso baseado em papéis (customer, support, admin)
Isolamento de tenant via row_filter
Proteção de PII via denied_fields
Todos os 4 modos de transporte (stdio, sse, streamable-http, websocket)

Consulte o README do exemplo para detalhes completos e consultas de exemplo.

_{MDBP — Model Database Protocol}

Model Database Protocol

Documentação

MDBP - Protocolo de Banco de Dados Modelo

Sumário

Instalação

Início Rápido

Funcionando em 3 Linhas

Exemplo de Saída

Resposta de Erro

Exemplo do Mundo Real (PostgreSQL)

Conexão de Banco de Dados Zero-Config

Conceitos Principais

O que é uma Intenção?

Pipeline

Tipos de Intenção

list - Listar Registros

get - Obter Registro Único

count - Contar Registros

aggregate - Agregar

create - Criar Registro

update - Atualizar Registro

delete - Excluir Registro

Filtragem

Filtros Simples (Sufixo de Operador)

Filtros Complexos (where)

Filtros de Subconsulta

Operações JOIN

JOIN Básico

Múltiplos JOINs

Self-JOIN (Alias)

Agregação

GROUP BY

HAVING

Modos Avançados de GROUP BY

Campos Calculados

CASE WHEN

Funções de Janela

Funções Escalares

CTE (Common Table Expressions)

Operações de Escrita

batch_create - Inserção em Massa

upsert - Inserir ou Atualizar

UPDATE com JOIN

RETURNING

Operações de Conjunto

UNION

Registro de Esquema

Auto-Descoberta (Padrão)

Registro Manual

Visualizando o Esquema

Mecanismo de Políticas

Definindo Políticas

Isolamento de Locatário

Restrição Global de Intenção

Consultando com um Papel

Mascaramento de Dados

Uso Básico

Estratégias Integradas

Opções de Estratégia (MaskingRule)

Funções de Mascaramento Personalizadas

Mascaramento + Campos Negados

Suporte a Arquivo de Configuração

Notas

Modo Dry-Run

Servidor MCP

Iniciando via CLI

Arquivo de Configuração

Integração com Claude Desktop

Uso Programático

Ferramentas MCP Expostas

Tratamento de Erros

Estrutura de Erro

Códigos de Erro

Erros de Esquema (MDBP_SCHEMA_*)

Erros de Política (MDBP_POLICY_*)

Erros de Intenção (MDBP_INTENT_*)

Erros de Consulta (MDBP_QUERY_*)

Erros de Conexão (MDBP_CONN_*)

Erros de Configuração

Tratamento de Erros no Código

Erros de Esquema (`MDBP_SCHEMA_*`)

Erros de Política (`MDBP_POLICY_*`)

Erros de Intenção (`MDBP_INTENT_*`)

Erros de Consulta (`MDBP_QUERY_*`)

Erros de Conexão (`MDBP_CONN_*`)