deep-db-agents¶

Simplified creation of Deep Agents (LangChain) —generic or specialized on a specific database (MySQL, MariaDB, Postgres, MongoDB, Neo4j, SQLite, DuckDB, Elasticsearch, OpenSearch)—through a single factory function.

📖 Full API reference: giurlanda.github.io/deep-db-agents

Idea¶

from deep_db_agents import create_deep_db_agents

agent = create_deep_db_agents(
    db_url="mysql://localhost:3306",
    credential={"user": "user", "password": "my_password", "database": "shop"},
    system="The `shop` database contains orders and customers. The orders table has millions of rows.",
    model="claude-sonnet-4-5-20250929",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "How many orders in 2025, by region?"}]
})

The factory:

reads the scheme of the URL (mysql, postgres, mongodb, neo4j, sqlite, duckdb…) to pick the dialect the agent should specialize on;
builds the tools suited to that database, injecting the credentials (which stay in the tools' closures, never in the prompt);
concatenates the dialect's generic system prompt — which encodes context-management principles — with the system prompt passed by the user;
delegates to create_deep_agent(tools=..., system_prompt=..., **kwargs).

All extra arguments (model, subagents, checkpointer, …) are forwarded as-is to create_deep_agent.

The library also lets you query several databases at once, through create_deep_db_multi_agents (see Multi-database agents), and offers a lighter, non-Deep-Agent alternative through create_db_agents (see create_db_agents: a lighter alternative).

Multi-database agents¶

create_deep_db_multi_agents builds an orchestrator agent that never queries a database directly: it delegates each sub-question to the sub-agent specialized on the relevant database (through the task tool) and combines the results. This is how you answer questions that span multiple data sources.

from deep_db_agents import create_deep_db_agents, create_deep_db_multi_agents

orders_agent = create_deep_db_agents(
    "postgres://localhost:5432",
    credential={"user": "reader", "password": "secret", "database": "orders"},
)
events_agent = create_deep_db_agents(
    "mongodb://localhost:27017",
    credential={"database": "events"},
)

orchestrator = create_deep_db_multi_agents(
    db_agents={
        "orders": {"description": "Orders and customers (Postgres)", "agent": orders_agent},
        "events": {"description": "Raw event log (MongoDB)", "agent": events_agent},
    },
    system="The two databases share the `customer_id` field; join results in memory.",
    model="claude-sonnet-4-5-20250929",
)

result = orchestrator.invoke(
    {"messages": [{"role": "user", "content": "Compare orders vs. events last week."}]}
)

Each sub-agent keeps its own tools and credentials in its own closures — the orchestrator only ever sees the sub-agents' descriptions and their final answers, never raw rows. db_agents values must be agents already built by create_deep_db_agents (compiled, with a working .invoke).

If sub-agents share a materialization backend (see below), register it once and pass the same be_uuid to every agent, including the orchestrator, and forward it through config on invoke/ainvoke — see the full example in Materializing results: the backend registry.

`create_db_agents`: a lighter alternative¶

create_db_agents builds a plain LangChain agent (langchain.agents.create_agent) instead of a Deep Agent. It goes through the same dialect resolution and tool/prompt construction as create_deep_db_agents, so credentials, guardrails and error feedback all behave identically — what differs is the surrounding harness:

No materialize_* tools. They require a deepagents filesystem backend to write to; the plain agent has none, so large-result materialization (see below) is unavailable — only the guardrail-limited run_query/sample_rows-style tools are exposed.
No Deep Agent scaffolding. No built-in planning/TodoList, no subagent delegation, no virtual filesystem — just a single agent with tools and a system prompt.
Same signature otherwise (db_url, credential, system, guardrails, metrics, **kwargs forwarded to create_agent).

Use it when the questions are simple enough that you don't need multi-step planning or file-backed results — e.g. quick lookups, dashboards, or a lightweight assistant embedded in another application.

from deep_db_agents import create_db_agents

agent = create_db_agents(
    "sqlite:///./data/app.db",
    system="Answer briefly, cite the exact table and column names.",
)
result = agent.invoke({"messages": [{"role": "user", "content": "List all tables."}]})

Code interpreter (experimental)¶

create_deep_db_agents(..., enable_code_interpreter=True) attaches a CodeInterpreterMiddleware (from the optional langchain-quickjs package, extra code-interpreter) to the agent. This gives the model a sandboxed JavaScript execution tool that can also call the dialect's own DB tools (ptc, "pass-through tools"), so it can fetch data and post-process it — reshape, aggregate, compute statistics — in one code-execution step instead of many separate tool calls, saving context.

pip install "deep-db-agents[code-interpreter]"        # from PyPI
uv pip install -e ".[code-interpreter]"                # from source

agent = create_deep_db_agents(
    "postgres://localhost:5432",
    credential={"user": "reader", "password": "secret", "database": "shop"},
    enable_code_interpreter=True,
)

This is experimental: the middleware and its interaction with the guardrails are newer and less battle-tested than the rest of the library, and its interface may still change. It does not bypass the guardrails (the code interpreter can only call the same wrapped tools the agent already has), but it does give the model a more general execution capability — enable it only if you need the extra data-manipulation power.

Context-management principles¶

The tools and generic prompts enforce this defense hierarchy:

aggregate in the DB → limit and paginate → explore before extracting → materialize to file → summarize → hard guardrails

Guardrails are enforced by the tool wrapper (not by the agent): a non-bypassable maximum LIMIT, query timeouts, row estimation via EXPLAIN, a SELECT-only whitelist, and a per-session row/token budget. Large datasets are materialized to file (Parquet/CSV), and only metadata and previews are passed back to the agent.

query_timeout_s is enforced as a client-side timeout in addition to the server-side one, so a query (or socket) that hangs gets interrupted instead of blocking the agent — especially when the model issues several tool calls in parallel. For SQLite/DuckDB, which lack a native statement_timeout, the limit is enforced by a watchdog calling interrupt().

Errors become feedback, not dead ends¶

When a query fails — bad syntax, a non-existent table/column/field, an incompatible operator or type — the driver exception is not propagated raw to interrupt the agent's turn. It is turned into a structured message (query_errors.format_query_error) that is returned as the tool's output: error type, driver detail, the offending query (truncated, and with any credentials in the message redacted), and a hint to fix and retry, exploring the schema first if needed. This lets the model self-correct within the same conversation instead of failing the whole run.

The same applies to whitelist/scope violations (e.g. a write statement, or an index outside the configured credential["index"] pattern on Elasticsearch/OpenSearch): the operation is blocked before reaching the driver, but reported back as corrective feedback rather than a hard error. Only the session-level guardrails (row budget exhausted, EXPLAIN row-estimate threshold exceeded) remain hard exceptions — those signal a limit the agent must not be allowed to work around.

Materializing results: the backend registry¶

The materialize_query tool (Deep Agent only, see create_db_agents: a lighter alternative) writes large results to a file (CSV or Parquet) and returns only metadata, a preview and numeric statistics to the agent — see MaterializedResult. To do that it needs a deepagents backend (a BackendProtocol implementation, e.g. FilesystemBackend) to write to, and the tool resolves it at call time, not at agent-construction time.

That indirection exists because the backend can't simply live in the tool's closure like the credentials do: the same backend is often shared across several agents (e.g. all sub-agents of a multi-database orchestrator), and passing the live object around risks leaking a reference that outlives the session. Instead:

Register the backend once in the process-wide BERegistry singleton — add() returns an opaque UUID.
Pass that UUID (not the backend) to every agent that should be able to use it, via config={"configurable": {"be_uuid": ...}} at invoke/ainvoke time.
Inside the tool, the dialect looks up runtime.config["configurable"]["be_uuid"] and resolves it back to the backend with BERegistry().get(be_uuid). If the key is missing or unregistered, the tool returns an error message to the agent instead of raising.
Remove the backend when you're done with the session (BERegistry().remove(be_uuid)) — whoever calls add() is responsible for the matching remove(), otherwise the instance stays referenced (and the workspace files reachable) for the whole process lifetime.

from deepagents.backends import FilesystemBackend
from deep_db_agents import create_deep_db_agents
from deep_db_agents.backend_registry import BERegistry

ber = BERegistry()
be_uuid = ber.add(FilesystemBackend(root_dir="./workspace", virtual_mode=True))

agent = create_deep_db_agents(
    "duckdb:///warehouse/dw.duckdb",
    backend=ber.get(be_uuid),  # forwarded to create_deep_agent, gives the agent its filesystem
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Export all 2025 orders to a file."}]},
    config={"configurable": {"thread_id": "session-1", "be_uuid": be_uuid}},
)

ber.remove(be_uuid)  # when the session ends

The backend= kwarg (forwarded to create_deep_agent) is what gives the agent its virtual filesystem in the first place; be_uuid in config is what lets the materialize_query tool find that same backend again when it runs. Both must point at the same registered instance.

Metrics¶

Pass a SessionMetrics instance (from deep_db_agents.observability) to metrics= on either factory function to get thread-safe counters for the whole session, readable after invoke:

queries_run / rows_returned — successful executions and total rows returned.
estimate_blocked — queries rejected by the EXPLAIN row-estimate guardrail before running.
budget_exhausted — times the per-session row budget was hit.

from deep_db_agents import create_deep_db_agents
from deep_db_agents.observability import SessionMetrics

metrics = SessionMetrics()
agent = create_deep_db_agents("postgres://localhost:5432", metrics=metrics, credential={...})
agent.invoke({"messages": [{"role": "user", "content": "How many orders last week?"}]})

print(metrics.summary())  # "queries run=3, rows returned=142, blocked by estimate=0, ..."

It's optional and created by the caller (one instance per agent/session) — the tools only update the counters, they never create or reset the object.

Installation¶

From PyPI¶

pip install "deep-db-agents[mysql,postgres,analysis]"   # install only the extras you need

From source (development)¶

uv venv
uv pip install -e ".[mysql,postgres,analysis,dev]"   # install only the extras you need

Available extras:

Extra	Installs	Purpose
`mysql`	`pymysql`	MySQL dialect driver
`mariadb`	`pymysql`	MariaDB dialect driver (reuses the MySQL driver)
`postgres`	`psycopg[binary]`	Postgres dialect driver
`mongodb`	`pymongo`	MongoDB dialect driver
`neo4j`	`neo4j`	Neo4j dialect driver
`duckdb`	`duckdb`	DuckDB dialect driver
`elasticsearch`	`elasticsearch`	Elasticsearch dialect driver
`opensearch`	`opensearch-py`	OpenSearch dialect driver
`analysis`	`pandas`, `pyarrow`	Parquet support for `materialize_query` (large-result materialization)
`code-interpreter`	`langchain-quickjs`	Sandboxed JS execution tool (see Code interpreter)
`all`	every extra above	Every dialect driver + `analysis` + `code-interpreter`
`dev`	`pytest`, `ruff`, `langchain-openai`, `rich`	Test/lint tooling for contributing to the library
`docs`	`mkdocs`, `mkdocs-material`, `mkdocstrings[python]`	Build the documentation site locally

SQLite needs no extra (it uses the stdlib sqlite3). Extras can be combined, e.g. pip install "deep-db-agents[postgres,mongodb,analysis]".

Dialect status¶

Database	URL scheme	Status
MySQL	`mysql`	Complete
MariaDB	`mariadb`	Complete
Postgres	`postgres`/`postgresql`	Complete
MongoDB	`mongodb`	Complete
Neo4j	`neo4j`	Complete
SQLite	`sqlite`	Complete
DuckDB	`duckdb`	Complete
Elasticsearch	`elasticsearch`	Complete
OpenSearch	`opensearch`	Complete

Credentials by dialect¶

host/port always come from the URL (<scheme>://host:port), never from credential; file- based dialects (SQLite, DuckDB) use the URL's path instead and read no connection credentials. Everything else — auth, target database, driver timeouts — is read from the credential dict, whose expected keys depend on the dialect:

Dialect	`credential` keys	Notes
MySQL / MariaDB	`user`, `password`, `database` (or `db`), `connect_timeout`, `read_timeout`	MariaDB reuses MySQL's connection logic unchanged.
Postgres	`user`, `password`, `database` (or `db`), `connect_timeout`	`database`/`db` maps to `dbname`.
MongoDB	`user`, `password`, `authSource` (or `auth_source`), `connect_timeout`	`database` selects the target DB (via `ConnectionConfig.database`).
Neo4j	`user`, `password`, `database`, `connect_timeout`	If `user` is omitted, the driver connects unauthenticated.
SQLite	`connect_timeout`	Path comes from the URL; always opened read-only (except `:memory:`).
DuckDB	(none)	Path comes from the URL; read-only unless `:memory:`; a folder path enables data-lake mode.
Elasticsearch	`use_ssl`, `verify_certs`, `ca_certs`, `api_key` or `user`/`password`, `index`	`api_key` takes priority over `user`/`password` if both are set.
OpenSearch	`use_ssl`, `verify_certs`, `ca_certs`, `user`/`password`, `index`	No `api_key` option, unlike Elasticsearch.

index (Elasticsearch/OpenSearch only) restricts the agent to a single index name, a CSV list, or a * wildcard pattern; every search tool validates the requested index against it and rejects out-of-scope access as corrective feedback, not a crash.

File-based databases (SQLite, DuckDB)¶

For file-based DBs, the URL carries a path instead of host:port, following the SQLAlchemy convention (the path is relative to the application's working directory):

create_deep_db_agents(db_url="sqlite:///data/app.db")      # relative:  ./data/app.db
create_deep_db_agents(db_url="sqlite:////var/lib/app.db")  # absolute:  /var/lib/app.db
create_deep_db_agents(db_url="duckdb:///warehouse/dw.duckdb")
create_deep_db_agents(db_url="duckdb:///lake/")            # data lake: see below

DuckDB data lake: if the path is a folder (trailing slash), the parquet/csv/json files inside it are exposed as queryable tables via SQL.
DuckDB files are opened read-only, so multiple parallel tool calls can connect to the same file.
:memory: is supported (sqlite://:memory:), but with SQLite it does not share state across tool calls (each call opens a new connection): suitable only for ephemeral/test usage.

Fully local example (local database + local LLM)¶

Nothing in the factory is Anthropic-specific: model/kwargs are forwarded as-is to create_deep_agent/create_agent, so any LangChain chat model works. Pairing a file-based dialect (no server to run) with a local model server (e.g. LM Studio or Ollama exposing an OpenAI-compatible endpoint) gives you a fully offline agent:

from langchain_openai import ChatOpenAI
from deep_db_agents import create_deep_db_agents

local_model = ChatOpenAI(
    model="qwen3-coder-30b",          # whatever model is loaded in LM Studio/Ollama
    base_url="http://127.0.0.1:1234/v1",
    api_key="not-needed",             # required by the client, ignored by the local server
    temperature=0.1,
)

agent = create_deep_db_agents(
    "sqlite:///./chinook.db",
    system="The `chinook` database is a digital music store (artists, albums, tracks, invoices).",
    model=local_model,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Which genre has the most tracks?"}]}
)

No credential is needed for a local SQLite file, no network egress is required for either the database or the model, and the same pattern works with DuckDB.

Development¶

ruff check src tests
pytest

Disclaimer¶

This library grants an LLM agent the ability to connect to and query real databases. While guardrails (statement whitelisting, timeouts, row limits, EXPLAIN thresholds) are enforced in code and are not bypassable by the agent's prompt, no safeguard eliminates all risk: model behavior can be unpredictable, and misconfiguration (e.g. overly broad credentials) is outside the library's control. Always point it at credentials scoped to the minimum required privileges, prefer read-only accounts for exploratory use, and test against non-production data before running it against anything that matters. Use of this library, and any consequences arising from it, is entirely at the user's own risk and responsibility.