API Reference¶

Auto-generated reference for the deep_db_agents package, extracted from source docstrings.

Public API¶

deep_db_agents ¶

deep-db-agents: factory for Deep Agents specialized on different databases.

GuardrailConfig `dataclass` ¶

GuardrailConfig(default_rows: int = 100, hard_max_rows: int = 1000, explain_row_threshold: int = 1000000, query_timeout_s: int = 30, allowed_statements: frozenset[str] = frozenset({'SELECT'}), row_budget: int | None = 50000)

Safety thresholds that cannot be bypassed by the agent.

Attributes:

Name	Type	Description
`default_rows`	`int`	LIMIT applied automatically when the agent does not specify one.
`hard_max_rows`	`int`	Hard cap on rows per single query, never exceedable.
`explain_row_threshold`	`int`	If the row estimate from `EXPLAIN` exceeds this threshold, the query is blocked and the agent is asked to refine its filters or aggregate instead.
`query_timeout_s`	`int`	Maximum execution timeout for a query, enforced by the driver/database.
`allowed_statements`	`frozenset[str]`	Whitelist of allowed statement types (only `SELECT` by default).
`row_budget`	`int \| None`	Cumulative row budget returned per session (`None` = unlimited).

check_estimate ¶

check_estimate(estimated_rows: int, metrics: SessionMetrics | None = None) -> None

Block the query if the EXPLAIN estimate exceeds the threshold.

Parameters:

Name	Type	Description	Default
`estimated_rows`	`int`	The estimated number of rows the query would return, typically obtained via `EXPLAIN`.	required
`metrics`	`SessionMetrics \| None`	Optional session counters to update when the query is blocked.	`None`

Raises:

Type	Description
`GuardrailError`	If `estimated_rows` exceeds `explain_row_threshold`.

Source code in src/deep_db_agents/guardrails.py

def check_estimate(self, estimated_rows: int, metrics: SessionMetrics | None = None) -> None:
    """Block the query if the EXPLAIN estimate exceeds the threshold.

    Args:
        estimated_rows: The estimated number of rows the query would return,
            typically obtained via ``EXPLAIN``.
        metrics: Optional session counters to update when the query is blocked.

    Raises:
        GuardrailError: If ``estimated_rows`` exceeds ``explain_row_threshold``.
    """
    if estimated_rows > self.explain_row_threshold:
        if metrics is not None:
            metrics.record_estimate_blocked()
        _logger.warning(
            "query blocked: estimate ~%d rows exceeds threshold %d",
            estimated_rows,
            self.explain_row_threshold,
        )
        raise GuardrailError(
            f"Query blocked: ~{estimated_rows:,} estimated rows exceed the "
            f"threshold of {self.explain_row_threshold:,}. Refine your filters "
            "or use aggregation."
        )

clamp_limit ¶

clamp_limit(requested: int | None) -> int

Compute the effective LIMIT to apply, capped by hard_max_rows.

Parameters:

Name	Type	Description	Default
`requested`	`int \| None`	The row limit requested by the agent, or `None`/non-positive to fall back to `default_rows`.	required

Returns:

Name	Type	Description
`int`	`int`	The effective limit, never greater than `hard_max_rows`.

Source code in src/deep_db_agents/guardrails.py

def clamp_limit(self, requested: int | None) -> int:
    """Compute the effective LIMIT to apply, capped by ``hard_max_rows``.

    Args:
        requested: The row limit requested by the agent, or ``None``/non-positive
            to fall back to ``default_rows``.

    Returns:
        int: The effective limit, never greater than ``hard_max_rows``.
    """
    if requested is None or requested <= 0:
        requested = self.default_rows
    return min(requested, self.hard_max_rows)

ConnectionConfig `dataclass` ¶

ConnectionConfig(scheme: str, host: str | None, port: int | None, credential: dict[str, Any] = dict(), path: str | None = None)

Connection parameters for a database.

credential is a free-form dictionary whose content depends on the specific DB (e.g. {"user": ..., "password": ...} or {"secret_key": ...}). Well-known keys (database/db) are exposed as convenience properties.

Attributes:

Name	Type	Description
`scheme`	`str`	The URL scheme identifying the dialect (e.g. `"postgres"`).
`host`	`str \| None`	Hostname for network databases, or `None` for file-based databases.
`port`	`int \| None`	Port for network databases, or `None` for file-based databases.
`credential`	`dict[str, Any]`	Free-form credential/connection dictionary.
`path`	`str \| None`	File/directory path for file-based databases (`sqlite`/`duckdb`); `None` for network databases. `:memory:` denotes an in-memory database.

database `property` ¶

database: str | None

Return the logical database/schema name, if provided in the credentials.

Returns:

Type	Description
`str \| None`	str \| None: The value of the `database` or `db` credential key, or
`str \| None`	`None` if neither is present.

repr ¶

__repr__() -> str

Return a repr with credentials masked.

Returns:

Name	Type	Description
`str`	`str`	A string representation of this config where `credential` content
	`str`	is masked so passwords never leak into logs, tracebacks, or error messages.

Source code in src/deep_db_agents/connection.py

def __repr__(self) -> str:
    """Return a repr with credentials masked.

    Returns:
        str: A string representation of this config where ``credential`` content
        is masked so passwords never leak into logs, tracebacks, or error messages.
    """
    # Mask the contents of ``credential`` so passwords never end up in logs,
    # tracebacks, or error messages when the object is printed.
    cred = "{…}" if self.credential else "{}"
    return (
        f"ConnectionConfig(scheme={self.scheme!r}, host={self.host!r}, "
        f"port={self.port!r}, credential={cred}, path={self.path!r})"
    )

SessionMetrics `dataclass` ¶

SessionMetrics(queries_run: int = 0, rows_returned: int = 0, budget_exhausted: int = 0, estimate_blocked: int = 0, _lock: Lock = threading.Lock())

Cumulative counters for a session/agent, updated by the guardrails.

Thread-safe: tool calls may run in parallel. Pass an instance to the factory (metrics parameter) and read it after invoking the agent.

Attributes:

Name	Type	Description
`queries_run`	`int`	Total number of queries executed.
`rows_returned`	`int`	Total number of rows returned across all queries.
`budget_exhausted`	`int`	Number of times the session row budget was exhausted.
`estimate_blocked`	`int`	Number of times a query was blocked by the EXPLAIN row-estimate threshold.

record_budget_exhausted ¶

record_budget_exhausted() -> None

Record that the session row budget was exhausted.

Returns:

Type	Description
`None`	None.

Source code in src/deep_db_agents/observability.py

def record_budget_exhausted(self) -> None:
    """Record that the session row budget was exhausted.

    Returns:
        None.
    """
    with self._lock:
        self.budget_exhausted += 1

record_estimate_blocked ¶

record_estimate_blocked() -> None

Record that a query was blocked by the EXPLAIN row-estimate threshold.

Returns:

Type	Description
`None`	None.

Source code in src/deep_db_agents/observability.py

def record_estimate_blocked(self) -> None:
    """Record that a query was blocked by the EXPLAIN row-estimate threshold.

    Returns:
        None.
    """
    with self._lock:
        self.estimate_blocked += 1

record_query ¶

record_query(rows: int) -> None

Record that a query ran and returned rows rows.

Parameters:

Name	Type	Description	Default
`rows`	`int`	Number of rows returned by the query.	required

Returns:

Type	Description
`None`	None.

Source code in src/deep_db_agents/observability.py

def record_query(self, rows: int) -> None:
    """Record that a query ran and returned ``rows`` rows.

    Args:
        rows: Number of rows returned by the query.

    Returns:
        None.
    """
    with self._lock:
        self.queries_run += 1
        self.rows_returned += rows

summary ¶

summary() -> str

Build a human-readable summary of the session counters.

Returns:

Name	Type	Description
`str`	`str`	A one-line summary of queries run, rows returned, and blocks.

Source code in src/deep_db_agents/observability.py

def summary(self) -> str:
    """Build a human-readable summary of the session counters.

    Returns:
        str: A one-line summary of queries run, rows returned, and blocks.
    """
    return (
        f"queries run={self.queries_run}, rows returned={self.rows_returned:,}, "
        f"blocked by estimate={self.estimate_blocked}, "
        f"budget exhausted={self.budget_exhausted}"
    )

create_deep_db_agents ¶

create_deep_db_agents(db_url: str, credential: dict[str, Any] | None = None, system: str = '', *, guardrails: GuardrailConfig | None = None, enable_code_interpreter: bool = False, metrics: SessionMetrics | None = None, **kwargs: Any)

Create a Deep Agent specialized on the database pointed to by db_url.

Parameters:

Name	Type	Description	Default
`db_url`	`str`	Database URL in the form `<scheme>://<host>:<port>`. The scheme (`mysql`, `postgres`, `mongodb`, `neo4j`, ...) selects the dialect.	required
`credential`	`dict[str, Any] \| None`	Access credentials; the expected keys depend on the target database (e.g. `{"user": ..., "password": ...}` or `{"secret_key": ...}`).	`None`
`system`	`str`	Database-specific system prompt, appended to the dialect's generic prompt.	`''`
`guardrails`	`GuardrailConfig \| None`	Safety thresholds for the tools (max LIMIT, timeout, EXPLAIN threshold, row budget). Defaults to `GuardrailConfig()` when omitted.	`None`
`enable_code_interpreter`	`bool`	If `True`, adds a `CodeInterpreterMiddleware` (requires the optional `code-interpreter` extra) that can call the generated DB tools.	`False`
`metrics`	`SessionMetrics \| None`	Optional `SessionMetrics`; when given, the tools update its session counters (queries run, rows returned, blocks from estimation and budget), readable after the call.	`None`
`**kwargs`	`Any`	Forwarded as-is to `create_deep_agent` (`model`, `subagents`, `checkpointer`, ...). Any `tools` passed here are merged with the dialect's tools.	`{}`

Returns:

Type	Description
	The compiled agent, with an `agent.invoke({"messages": [...]}, config=...)`
	interface. Also supports `await agent.ainvoke(...)`: tools are synchronous, but
	under `ainvoke` LangChain runs them in a thread pool, dispatching a turn's tool
	calls concurrently (see `examples/async_quickstart.py`).

Example

from deep_db_agents import create_deep_db_agents

agent = create_deep_db_agents(
    "postgres://localhost:5432/mydb",
    credential={"user": "reader", "password": "secret"},
    system="Focus on the `orders` and `customers` tables.",
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "How many orders last week?"}]}
)

Source code in src/deep_db_agents/factory.py

def create_deep_db_agents(
    db_url: str,
    credential: dict[str, Any] | None = None,
    system: str = "",
    *,
    guardrails: GuardrailConfig | None = None,
    enable_code_interpreter: bool = False,
    metrics: SessionMetrics | None = None,
    **kwargs: Any,
):
    """Create a Deep Agent specialized on the database pointed to by ``db_url``.

    Args:
        db_url: Database URL in the form ``<scheme>://<host>:<port>``. The scheme
            (``mysql``, ``postgres``, ``mongodb``, ``neo4j``, ...) selects the dialect.
        credential: Access credentials; the expected keys depend on the target database
            (e.g. ``{"user": ..., "password": ...}`` or ``{"secret_key": ...}``).
        system: Database-specific system prompt, appended to the dialect's generic prompt.
        guardrails: Safety thresholds for the tools (max LIMIT, timeout, EXPLAIN threshold,
            row budget). Defaults to ``GuardrailConfig()`` when omitted.
        enable_code_interpreter: If ``True``, adds a ``CodeInterpreterMiddleware`` (requires
            the optional ``code-interpreter`` extra) that can call the generated DB tools.
        metrics: Optional ``SessionMetrics``; when given, the tools update its session
            counters (queries run, rows returned, blocks from estimation and budget),
            readable after the call.
        **kwargs: Forwarded as-is to ``create_deep_agent`` (``model``, ``subagents``,
            ``checkpointer``, ...). Any ``tools`` passed here are merged with the dialect's
            tools.

    Returns:
        The compiled agent, with an ``agent.invoke({"messages": [...]}, config=...)``
        interface. Also supports ``await agent.ainvoke(...)``: tools are synchronous, but
        under ``ainvoke`` LangChain runs them in a thread pool, dispatching a turn's tool
        calls concurrently (see ``examples/async_quickstart.py``).

    Example:
        ```python
        from deep_db_agents import create_deep_db_agents

        agent = create_deep_db_agents(
            "postgres://localhost:5432/mydb",
            credential={"user": "reader", "password": "secret"},
            system="Focus on the `orders` and `customers` tables.",
        )
        result = agent.invoke(
            {"messages": [{"role": "user", "content": "How many orders last week?"}]}
        )
        ```
    """
    db_tools, prompt = _build_dialect_parts(
        db_url, credential, system, guardrails, materialize_enable=True, metrics=metrics
    )
    user_tools = kwargs.pop("tools", []) or []

    if enable_code_interpreter:
        # Lazy import: langchain-quickjs is an optional extra (``code-interpreter``).
        from langchain_quickjs import CodeInterpreterMiddleware

        code_interpreter = CodeInterpreterMiddleware(ptc=[*db_tools, *user_tools])
        # Accepts any middleware sequence (list, tuple...) without overwriting it.
        existing = list(kwargs.pop("middleware", []) or [])
        kwargs["middleware"] = [*existing, code_interpreter]

    return create_deep_agent(
        tools=[*db_tools, *user_tools],
        system_prompt=prompt,
        **kwargs,
    )

create_deep_db_multi_agents ¶

create_deep_db_multi_agents(db_agents: Mapping[str, Mapping[str, Any]], system: str = '', **kwargs: Any)

Create a Deep Agent orchestrator that coordinates multiple deep_db_agents.

The orchestrator never queries a database directly: it delegates each sub-question to the sub-agent specialized on the relevant database (exposed through the task tool) and combines the results. This enables answering questions that span multiple databases.

Parameters:

Name	Type	Description	Default
`db_agents`	`Mapping[str, Mapping[str, Any]]`	Mapping of `name -> {"description": ..., "agent": ...}`. `description` is a short description of the sub-agent and its database (the orchestrator uses it to decide whom to delegate to); `agent` is an already-compiled agent from :func:`create_deep_db_agents`.	required
`system`	`str`	Extra system prompt, appended to the orchestrator's generic prompt and the sub-agent roster, to provide additional context (domain, join rules, ...).	`''`
`**kwargs`	`Any`	Forwarded as-is to `create_deep_agent` (`model`, `middleware`, `checkpointer`, ...). Any `subagents` passed here are merged with the ones derived from `db_agents`.	`{}`

Returns:

Type	Description
	The compiled orchestrator agent, with an `agent.invoke({"messages": [...]})`
	interface.

Raises:

Type	Description
`InvalidMultiAgentConfigError`	If `db_agents` is empty, an entry is missing the `description`/`agent` keys, or `agent` is not a compiled agent (no `.invoke`).

Example

from deep_db_agents import create_deep_db_agents, create_deep_db_multi_agents

orders_agent = create_deep_db_agents("postgres://localhost:5432/orders")
events_agent = create_deep_db_agents("mongodb://localhost:27017/events")

orchestrator = create_deep_db_multi_agents(
    {
        "orders": {
            "description": "Orders and customers (Postgres)",
            "agent": orders_agent,
        },
        "events": {"description": "Raw event log (MongoDB)", "agent": events_agent},
    }
)
query = "Compare orders vs. events last week."
result = orchestrator.invoke({"messages": [{"role": "user", "content": query}]})

Source code in src/deep_db_agents/factory.py

def create_deep_db_multi_agents(
    db_agents: Mapping[str, Mapping[str, Any]],
    system: str = "",
    **kwargs: Any,
):
    """Create a Deep Agent orchestrator that coordinates multiple ``deep_db_agents``.

    The orchestrator never queries a database directly: it delegates each sub-question to
    the sub-agent specialized on the relevant database (exposed through the ``task`` tool)
    and combines the results. This enables answering questions that span multiple databases.

    Args:
        db_agents: Mapping of ``name -> {"description": ..., "agent": ...}``. ``description``
            is a short description of the sub-agent and its database (the orchestrator uses
            it to decide whom to delegate to); ``agent`` is an already-compiled agent from
            :func:`create_deep_db_agents`.
        system: Extra system prompt, appended to the orchestrator's generic prompt and the
            sub-agent roster, to provide additional context (domain, join rules, ...).
        **kwargs: Forwarded as-is to ``create_deep_agent`` (``model``, ``middleware``,
            ``checkpointer``, ...). Any ``subagents`` passed here are merged with the ones
            derived from ``db_agents``.

    Returns:
        The compiled orchestrator agent, with an ``agent.invoke({"messages": [...]})``
        interface.

    Raises:
        InvalidMultiAgentConfigError: If ``db_agents`` is empty, an entry is missing the
            ``description``/``agent`` keys, or ``agent`` is not a compiled agent (no
            ``.invoke``).

    Example:
        ```python
        from deep_db_agents import create_deep_db_agents, create_deep_db_multi_agents

        orders_agent = create_deep_db_agents("postgres://localhost:5432/orders")
        events_agent = create_deep_db_agents("mongodb://localhost:27017/events")

        orchestrator = create_deep_db_multi_agents(
            {
                "orders": {
                    "description": "Orders and customers (Postgres)",
                    "agent": orders_agent,
                },
                "events": {"description": "Raw event log (MongoDB)", "agent": events_agent},
            }
        )
        query = "Compare orders vs. events last week."
        result = orchestrator.invoke({"messages": [{"role": "user", "content": query}]})
        ```
    """
    if not db_agents:
        raise InvalidMultiAgentConfigError("db_agents must not be empty.")

    subagents: list[CompiledSubAgent] = []
    roster_lines: list[str] = []
    for name, spec in db_agents.items():
        if not isinstance(spec, Mapping) or "description" not in spec or "agent" not in spec:
            raise InvalidMultiAgentConfigError(
                f"db_agents[{name!r}] must be a dict with the 'description' and 'agent' keys."
            )
        if not callable(getattr(spec["agent"], "invoke", None)):
            raise InvalidMultiAgentConfigError(
                f"db_agents[{name!r}]['agent'] must be a compiled agent (with .invoke), "
                f"got {type(spec['agent']).__name__!r}."
            )
        description = spec["description"]
        # CompiledSubAgent: the agent is already a compiled runnable, used as-is.
        subagents.append(
            CompiledSubAgent(name=name, description=description, runnable=spec["agent"])
        )
        roster_lines.append(f"- `{name}`: {description}")

    prompt = ORCHESTRATOR_SYSTEM_PROMPT + _ORCHESTRATOR_ROSTER_HEADER + "\n".join(roster_lines)
    if system:
        prompt += _ORCHESTRATOR_INSTRUCTIONS_HEADER + system

    extra_subagents = kwargs.pop("subagents", []) or []
    return create_deep_agent(
        system_prompt=prompt,
        subagents=[*subagents, *extra_subagents],
        **kwargs,
    )

create_db_agents ¶

create_db_agents(db_url: str, credential: dict[str, Any] | None = None, system: str = '', *, guardrails: GuardrailConfig | None = None, metrics: SessionMetrics | None = None, **kwargs: Any)

Create a plain LangChain Agent specialized on the database pointed to by db_url.

Unlike :func:create_deep_db_agents, the file-materialization tools (materialize_*) are not exposed: they require the deepagents filesystem backend, which the plain agent does not have.

Parameters:

Name	Type	Description	Default
`db_url`	`str`	Database URL in the form `<scheme>://<host>:<port>`. The scheme (`mysql`, `postgres`, `mongodb`, `neo4j`, ...) selects the dialect.	required
`credential`	`dict[str, Any] \| None`	Access credentials; the expected keys depend on the target database (e.g. `{"user": ..., "password": ...}` or `{"secret_key": ...}`).	`None`
`system`	`str`	Database-specific system prompt, appended to the dialect's generic prompt.	`''`
`guardrails`	`GuardrailConfig \| None`	Safety thresholds for the tools (max LIMIT, timeout, EXPLAIN threshold, row budget). Defaults to `GuardrailConfig()` when omitted.	`None`
`metrics`	`SessionMetrics \| None`	Optional `SessionMetrics`; when given, the tools update its session counters (queries run, rows returned, blocks from estimation and budget), readable after the call.	`None`
`**kwargs`	`Any`	Forwarded as-is to `create_agent` (`model`, `checkpointer`, ...). Any `tools` passed here are merged with the dialect's tools.	`{}`

Returns:

Type	Description
	The compiled agent, with an `agent.invoke({"messages": [...]}, config=...)`
	interface. Also supports `await agent.ainvoke(...)` (synchronous tools run in a
	thread pool).

Example

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

Source code in src/deep_db_agents/factory.py

def create_db_agents(
    db_url: str,
    credential: dict[str, Any] | None = None,
    system: str = "",
    *,
    guardrails: GuardrailConfig | None = None,
    metrics: SessionMetrics | None = None,
    **kwargs: Any,
):
    """Create a plain LangChain Agent specialized on the database pointed to by ``db_url``.

    Unlike :func:`create_deep_db_agents`, the file-materialization tools (``materialize_*``)
    are not exposed: they require the deepagents filesystem backend, which the plain agent
    does not have.

    Args:
        db_url: Database URL in the form ``<scheme>://<host>:<port>``. The scheme
            (``mysql``, ``postgres``, ``mongodb``, ``neo4j``, ...) selects the dialect.
        credential: Access credentials; the expected keys depend on the target database
            (e.g. ``{"user": ..., "password": ...}`` or ``{"secret_key": ...}``).
        system: Database-specific system prompt, appended to the dialect's generic prompt.
        guardrails: Safety thresholds for the tools (max LIMIT, timeout, EXPLAIN threshold,
            row budget). Defaults to ``GuardrailConfig()`` when omitted.
        metrics: Optional ``SessionMetrics``; when given, the tools update its session
            counters (queries run, rows returned, blocks from estimation and budget),
            readable after the call.
        **kwargs: Forwarded as-is to ``create_agent`` (``model``, ``checkpointer``, ...).
            Any ``tools`` passed here are merged with the dialect's tools.

    Returns:
        The compiled agent, with an ``agent.invoke({"messages": [...]}, config=...)``
        interface. Also supports ``await agent.ainvoke(...)`` (synchronous tools run in a
        thread pool).

    Example:
        ```python
        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."}]})
        ```
    """
    # Unlike create_deep_db_agents, materialization tools stay excluded: the plain
    # LangChain agent has no deepagents filesystem/backend to save them to.
    db_tools, prompt = _build_dialect_parts(
        db_url, credential, system, guardrails, materialize_enable=False, metrics=metrics
    )

    user_tools = kwargs.pop("tools", []) or []
    return create_agent(
        tools=[*db_tools, *user_tools],
        system_prompt=prompt,
        **kwargs,
    )

configure_logging ¶

configure_logging(level: int = logging.INFO) -> None

Attach a stderr handler to the library logger (convenience, idempotent).

Parameters:

Name	Type	Description	Default
`level`	`int`	Logging level to set on the library logger. Defaults to `logging.INFO`.	`INFO`

Returns:

Type	Description
`None`	None.

Source code in src/deep_db_agents/observability.py

def configure_logging(level: int = logging.INFO) -> None:
    """Attach a ``stderr`` handler to the library logger (convenience, idempotent).

    Args:
        level: Logging level to set on the library logger. Defaults to
            ``logging.INFO``.

    Returns:
        None.
    """
    logger = logging.getLogger(LOGGER_NAMESPACE)
    if not any(getattr(h, "_deep_db_agents", False) for h in logger.handlers):
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
        handler._deep_db_agents = True  # type: ignore[attr-defined]
        logger.addHandler(handler)
    logger.setLevel(level)

available_schemes ¶

available_schemes() -> list[str]

List the schemes currently registered.

Returns:

Type	Description
`list[str]`	list[str]: The list of registered scheme names.

Source code in src/deep_db_agents/registry.py

def available_schemes() -> list[str]:
    """List the schemes currently registered.

    Returns:
        list[str]: The list of registered scheme names.
    """
    return list(_REGISTRY)

API Reference¶

Public API¶

deep_db_agents ¶

GuardrailConfig dataclass ¶

check_estimate ¶

clamp_limit ¶

ConnectionConfig dataclass ¶

database property ¶

__repr__ ¶

SessionMetrics dataclass ¶

record_budget_exhausted ¶

record_estimate_blocked ¶

record_query ¶

summary ¶

create_deep_db_agents ¶

create_deep_db_multi_agents ¶

create_db_agents ¶

configure_logging ¶

available_schemes ¶

GuardrailConfig `dataclass` ¶

ConnectionConfig `dataclass` ¶

database `property` ¶

repr ¶

SessionMetrics `dataclass` ¶