Database¶

Database — mvp.database

Cluster: Core Infrastructure | Type: component | MCP Tools: None

Overview¶

SQL database abstraction with a pluggable adapter pattern, shipping a SQLite adapter for both in-memory and file-backed databases with full query execution, schema introspection, machine-readable adapter capabilities, and a 31-op MCP sub-package.

When to use:

Persisting structured data with SQL queries
Running ad-hoc SQL against in-memory or file-backed SQLite stores
Providing database backends for MCP sub-packages across components
Discovering backend-native operations, pooling semantics, migration dialects, unsupported features, and degraded capability envelopes before executing work

Example:

from mvp.database import DatabaseBlock, DBQuery

block = DatabaseBlock(name="db")
result = block.infer(DBQuery(sql="SELECT 1 AS value"))
# result.ok → True; result.value → DBResult with rows=[{"value": 1}]

MCP degraded responses surface completion_state, warning_card, and evidence. Runtime capability checks return verified; recoverable gaps such as ERD renderer absence or ignored per-request backend fields return qualified-draft; unavailable configured adapters return blocked-escalated.

Works well with: adapt_pandas, adapt_healing, goal_engine

Public API¶

`DatabaseAdapter(ABC)`¶

Contract that every backend adapter must satisfy.

Constructor:

Parameter	Type	Default
`config`	`DBConfig`	required

Methods:

`execute(query: DBQuery) -> Result[DBResult]`¶

Run a query/mutation and return a Result[DBResult].

`execute_transaction(statements: list[str]) -> Result[DBResult]`¶

Execute multiple statements atomically with rollback on failure.

`list_tables() -> list[dict]`¶

Return [{"name": ..., "row_count": ...}, ...].

`describe_table(name: str) -> list[dict]`¶

Return column metadata: [{"name", "type", "notnull", "pk"}, ...].

`get_schema() -> str`¶

Return DDL dump (or equivalent) for the entire database.

`info() -> str`¶

Return a short human-readable status string.

`close() -> None`¶

Release resources. Default is a no-op.

`dialect() -> str`¶

Return the backend dialect name used for SQL/migration semantics.

`capabilities() -> dict[str, Any]`¶

Return machine-readable backend capabilities.

`constraints() -> dict[str, Any]`¶

Return machine-readable safety and configuration constraints.

`health() -> dict[str, Any]`¶

Health envelope. Performs a real liveness probe where one is available

`supports(operation: str) -> bool`¶

Return whether the adapter declares support for an operation.

`DatabaseBlock(AIBlock[DBQuery, DBResult, Any])`¶

Executes database queries against a configured backend.

Field	Type	Default
`name`	`str`	'database'
`db_config`	`DBConfig`	field(default_factory=DBConfig)
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: DBQuery | RelationalQuery | RedisCommand | MongoCommand | Neo4jCommand) -> Result[DBResult]`¶

`DBConfig(BaseModel)`¶

Connection configuration for a database backend.

Field	Type	Default
`backend`	`Literal['sqlite', 'postgresql', 'redis', 'mongodb', 'neo4j']`	'sqlite'
`url`	`str`	':memory:'
`database`	`str`	''
`options`	`dict[str, Any]`	Field(default_factory=dict)
`pool_size`	`int`	5
`timeout_sec`	`int`	30

`DBQuery(BaseModel)`¶

A SQL or query to execute against the configured database.

Field	Type	Default
`sql`	`str`	required
`params`	`list[Any]`	Field(default_factory=list)
`key`	`str`	''
`document`	`dict[str, Any]`	Field(default_factory=dict)
`collection`	`str`	''
`operation`	`str`	''
`native_op`	`str`	''

`RelationalQuery(BaseModel)`¶

Typed relational SQL operation for SQLite/PostgreSQL callers.

Field	Type	Default
`sql`	`str`	required
`params`	`list[Any]`	Field(default_factory=list)
`operation`	`Literal['query', 'execute', 'transaction']`	'query'

`RedisCommand(BaseModel)`¶

Typed Redis key/value or pipeline operation.

Field	Type	Default
`operation`	`Literal['get', 'set', 'delete', 'exists', 'keys', 'scan', 'incr', 'pipeline']`	required
`key`	`str`	''
`document`	`dict[str, Any]`	Field(default_factory=dict)

`MongoCommand(BaseModel)`¶

Typed MongoDB collection/document operation.

Field	Type	Default
`operation`	`Literal['insert', 'insert_many', 'find', 'query', 'find_one', 'update', 'delete', 'count', 'aggregate']`	required
`collection`	`str`	''
`key`	`str`	''
`document`	`dict[str, Any]`	Field(default_factory=dict)

`Neo4jCommand(BaseModel)`¶

Typed Neo4j Cypher operation.

Field	Type	Default
`cypher`	`str`	required
`params`	`dict[str, Any]`	Field(default_factory=dict)
`operation`	`Literal['cypher']`	'cypher'

`DBResult(BaseModel)`¶

Result of a database operation.

Field	Type	Default
`rows`	`list[dict[str, Any]]`	Field(default_factory=list)
`affected_rows`	`int`	0
`columns`	`list[str]`	Field(default_factory=list)
`error`	`str`	''
`success`	`bool`	True
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`SqliteAdapter(DatabaseAdapter)`¶

sqlite3-backed database adapter.

Constructor:

Parameter	Type	Default
`config`	`DBConfig`	required

Methods:

`execute(query: DBQuery) -> Result[DBResult]`¶

`execute_transaction(statements: list[str]) -> Result[DBResult]`¶

`list_tables() -> list[dict]`¶

`describe_table(name: str) -> list[dict]`¶

`get_schema() -> str`¶

`info() -> str`¶

`dialect() -> str`¶

`capabilities() -> dict[str, Any]`¶

`constraints() -> dict[str, Any]`¶

`close() -> None`¶

`AdaptDatabaseMCPBlock(AIBlock[MCPDatabaseInput, MCPDatabaseOutput, dict])`¶

31-op database MCP block: core DB, schema, DDD entities, ERD,

Field	Type	Default
`name`	`str`	'adapt_database_mcp'
`state`	`dict \\| None`	None
`db_path`	`str`	':memory:'

Methods:

`infer(inp: MCPDatabaseInput) -> Result[MCPDatabaseOutput]`¶

`MCPDatabaseInput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`sql`	`str`	''
`params_json`	`str`	'[]'
`table_name`	`str`	''
`name`	`str`	''
`fields_json`	`str`	'[]'
`relationships_json`	`str`	'[]'
`description`	`str`	''
`tags_json`	`str`	'[]'
`nl_prompt`	`str`	''
`format`	`str`	'dot'
`output_path`	`str`	''
`schema_a`	`str`	''
`schema_b`	`str`	''
`version`	`str`	''
`migration_sql`	`str`	''
`rollback_sql`	`str`	''
`dry_run`	`bool`	False
`n`	`int`	1
`query`	`str`	''
`notes`	`str`	''
`backend`	`str`	'sqlite'
`connection_string`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPDatabaseOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`ok`	`bool`	True
`data`	`list[dict[str, Any]]`	Field(default_factory=list)
`columns`	`list[str]`	Field(default_factory=list)
`text`	`str`	''
`schema_sql`	`str`	''
`dot_src`	`str`	''
`entities`	`list[dict[str, Any]]`	Field(default_factory=list)
`migrations`	`list[dict[str, Any]]`	Field(default_factory=list)
`sessions`	`list[dict[str, Any]]`	Field(default_factory=list)
`error`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`DatabaseMCPStore`¶

Sync SQLite store with 5 tables for database MCP persistence.

Constructor:

Parameter	Type	Default
`db_path`	`str \\| None`	None

Methods:

Database¶

Overview¶

Public API¶

DatabaseAdapter(ABC)¶

execute(query: DBQuery) -> Result[DBResult]¶

execute_transaction(statements: list[str]) -> Result[DBResult]¶

list_tables() -> list[dict]¶

describe_table(name: str) -> list[dict]¶

get_schema() -> str¶

info() -> str¶

close() -> None¶

dialect() -> str¶

capabilities() -> dict[str, Any]¶

constraints() -> dict[str, Any]¶

health() -> dict[str, Any]¶

supports(operation: str) -> bool¶

DatabaseBlock(AIBlock[DBQuery, DBResult, Any])¶

infer(data: DBQuery | RelationalQuery | RedisCommand | MongoCommand | Neo4jCommand) -> Result[DBResult]¶

DBConfig(BaseModel)¶

DBQuery(BaseModel)¶

RelationalQuery(BaseModel)¶

RedisCommand(BaseModel)¶

MongoCommand(BaseModel)¶

Neo4jCommand(BaseModel)¶

DBResult(BaseModel)¶

SqliteAdapter(DatabaseAdapter)¶

execute(query: DBQuery) -> Result[DBResult]¶

execute_transaction(statements: list[str]) -> Result[DBResult]¶

list_tables() -> list[dict]¶

describe_table(name: str) -> list[dict]¶

get_schema() -> str¶

info() -> str¶

dialect() -> str¶

capabilities() -> dict[str, Any]¶

constraints() -> dict[str, Any]¶

close() -> None¶

AdaptDatabaseMCPBlock(AIBlock[MCPDatabaseInput, MCPDatabaseOutput, dict])¶

infer(inp: MCPDatabaseInput) -> Result[MCPDatabaseOutput]¶

MCPDatabaseInput(BaseModel)¶

MCPDatabaseOutput(BaseModel)¶

DatabaseMCPStore¶

upsert_entity(name: str, fields_json: str, relationships_json: str, description: str, tags: str) -> str¶

get_entity(name: str) -> dict[str, Any] | None¶

list_entities() -> list[dict[str, Any]]¶

delete_entity(name: str) -> bool¶

save_session(name: str, entities_json: str, schema_json: str, notes: str) -> str¶

load_session(name: str) -> dict[str, Any] | None¶

list_sessions() -> list[dict[str, Any]]¶

add_migration(version: str, description: str, migration_sql: str, rollback_sql: str) -> str¶

mark_applied(migration_id: str) -> None¶

list_migrations() -> list[dict[str, Any]]¶

get_pending_migrations() -> list[dict[str, Any]]¶

get_applied_migrations(n: int) -> list[dict[str, Any]]¶

get_erd_cache(source_hash: str, format: str) -> dict[str, Any] | None¶

set_erd_cache(source_hash: str, format: str, output_path: str, output_dot: str) -> str¶

log_llm(op: str, input_json: str, output_text: str) -> str¶

count_all() -> dict[str, int]¶

`DatabaseAdapter(ABC)`¶

`execute(query: DBQuery) -> Result[DBResult]`¶

`execute_transaction(statements: list[str]) -> Result[DBResult]`¶

`list_tables() -> list[dict]`¶

`describe_table(name: str) -> list[dict]`¶

`get_schema() -> str`¶

`info() -> str`¶

`close() -> None`¶

`dialect() -> str`¶

`capabilities() -> dict[str, Any]`¶

`constraints() -> dict[str, Any]`¶

`health() -> dict[str, Any]`¶

`supports(operation: str) -> bool`¶

`DatabaseBlock(AIBlock[DBQuery, DBResult, Any])`¶

`infer(data: DBQuery | RelationalQuery | RedisCommand | MongoCommand | Neo4jCommand) -> Result[DBResult]`¶

`DBConfig(BaseModel)`¶

`DBQuery(BaseModel)`¶

`RelationalQuery(BaseModel)`¶

`RedisCommand(BaseModel)`¶

`MongoCommand(BaseModel)`¶

`Neo4jCommand(BaseModel)`¶

`DBResult(BaseModel)`¶

`SqliteAdapter(DatabaseAdapter)`¶

`execute(query: DBQuery) -> Result[DBResult]`¶

`execute_transaction(statements: list[str]) -> Result[DBResult]`¶

`list_tables() -> list[dict]`¶

`describe_table(name: str) -> list[dict]`¶

`get_schema() -> str`¶

`info() -> str`¶

`dialect() -> str`¶

`capabilities() -> dict[str, Any]`¶

`constraints() -> dict[str, Any]`¶

`close() -> None`¶

`AdaptDatabaseMCPBlock(AIBlock[MCPDatabaseInput, MCPDatabaseOutput, dict])`¶

`infer(inp: MCPDatabaseInput) -> Result[MCPDatabaseOutput]`¶

`MCPDatabaseInput(BaseModel)`¶

`MCPDatabaseOutput(BaseModel)`¶

`DatabaseMCPStore`¶

`upsert_entity(name: str, fields_json: str, relationships_json: str, description: str, tags: str) -> str`¶

`get_entity(name: str) -> dict[str, Any] | None`¶

`list_entities() -> list[dict[str, Any]]`¶

`delete_entity(name: str) -> bool`¶

`save_session(name: str, entities_json: str, schema_json: str, notes: str) -> str`¶

`load_session(name: str) -> dict[str, Any] | None`¶

`list_sessions() -> list[dict[str, Any]]`¶

`add_migration(version: str, description: str, migration_sql: str, rollback_sql: str) -> str`¶

`mark_applied(migration_id: str) -> None`¶

`list_migrations() -> list[dict[str, Any]]`¶

`get_pending_migrations() -> list[dict[str, Any]]`¶

`get_applied_migrations(n: int) -> list[dict[str, Any]]`¶

`get_erd_cache(source_hash: str, format: str) -> dict[str, Any] | None`¶

`set_erd_cache(source_hash: str, format: str, output_path: str, output_dot: str) -> str`¶

`log_llm(op: str, input_json: str, output_text: str) -> str`¶

`count_all() -> dict[str, int]`¶