Advanced

Per-tool model overrides

You can run specific internal tools (for example a visualization or summarization tool) on a different model than the main agent. Use tool_overrides, mapping a tool name to a ModelOverides:

from sqlsaber import SQLSaber, SQLSaberOptions, ModelOverides

options = SQLSaberOptions(
    database="sqlite:///my.db",
    model_name="anthropic:claude-sonnet-4-5-20250929",
    tool_overrides={
        "viz": ModelOverides(
            model_name="openai:gpt-5-mini",
            api_key="sk-...",
        ),
    },
)

You may also pass a plain dict instead of ModelOverides:

tool_overrides={"viz": {"model_name": "openai:gpt-5-mini"}}

Note

An api_key override requires a model_name, since the provider is derived from the model. Setting api_key without model_name raises an error.

Custom system prompt

Override the agent’s system prompt with inline text or a path to a file:

from pathlib import Path

# Inline
SQLSaberOptions(
    database="sqlite:///my.db",
    system_prompt="You are a senior data analyst. Prefer concise answers.",
)

# From a file
SQLSaberOptions(
    database="sqlite:///my.db",
    system_prompt=Path("prompts/analyst.txt"),
)

Allowing write operations

By default the SDK only permits read-only SELECT queries. Set allow_dangerous=True to permit DML and a restricted subset of DDL:

SQLSaberOptions(database="sqlite:///my.db", allow_dangerous=True)

Caution

This allows data modifications (INSERT/UPDATE/DELETE, with UPDATE/DELETE requiring a WHERE) and limited DDL. Destructive operations such as DROP and TRUNCATE remain blocked. See Running Queries for the full policy.

Injecting a knowledge base

Provide domain context by injecting your own KnowledgeManager. The agent can search it while answering:

import asyncio

from sqlsaber import SQLSaber, SQLSaberOptions
from sqlsaber.knowledge import KnowledgeManager, SQLiteKnowledgeStore


async def main() -> None:
    manager = KnowledgeManager(store=SQLiteKnowledgeStore(db_path="knowledge.db"))

    await manager.add_knowledge(
        database_name="analytics",
        name="Gross margin",
        description="(Revenue - COGS) / Revenue",
        source="finance-handbook",
    )

    async with SQLSaber(
        options=SQLSaberOptions(
            database="analytics",
            knowledge_manager=manager,
        )
    ) as saber:
        print(await saber.query("How do we define gross margin?"))


if __name__ == "__main__":
    asyncio.run(main())

Note

When you inject your own knowledge_manager, the session does not close it for you — you own its lifecycle. If you omit it, the session creates and closes one automatically.

See the Knowledge Base guide for managing entries.

Persisting conversation threads

Inject a ThreadManager to persist each completed query’s history, as the CLI does between sessions:

from sqlsaber import SQLSaber, SQLSaberOptions
from sqlsaber.threads.manager import ThreadManager

options = SQLSaberOptions(
    database="analytics",
    thread_manager=ThreadManager(),
)

async with SQLSaber(options=options) as saber:
    result = await saber.query("How many orders shipped last week?")

The session saves each completed query and ends the current thread when it closes. The default manager uses SQLsaber’s local thread storage. Inject a manager with your own ThreadStorage when the application needs a different storage location or lifecycle.

Resume a stored thread with SQLSaber.resume(). If the thread stores a configured database name, leave options.database as None and SQLsaber selects that name:

import asyncio

from sqlsaber import SQLSaber, SQLSaberOptions


async def continue_thread() -> None:
    options = SQLSaberOptions(database=None)
    saber = await SQLSaber.resume("bb7b4d72", options=options)
    try:
        result = await saber.query("Now compare that with last quarter")
        print(result.text)
    finally:
        await saber.close()


asyncio.run(continue_thread())

Pass an explicit options.database to override the stored selector. SQLsaber does not persist raw connection strings or file paths for automatic resume.

SQLSaber.resume() raises typed errors. Catch ThreadNotFoundError when the ID does not exist, ThreadResumeHistoryError when the stored Pydantic AI history is missing or corrupt, ThreadDatabaseRequiredError when a safe database selector is missing, and ThreadDatabaseUnavailableError when a stored configured database is no longer available. All resume errors inherit from ThreadResumeError.

See the Conversation Threads guide for CLI thread commands.