Advanced

Capability models

Pass model and credential settings directly to the capability factory. Use functools.partial to bind those settings before SQLsaber supplies the session context:

from functools import partial

from sqlsaber import SQLSaber, SQLSaberOptions
from sqlsaber_viz import capability as visualization

options = SQLSaberOptions(
    database="sqlite:///my.db",
    model_name="anthropic:claude-sonnet-4-5-20250929",
    capabilities=[
        partial(
            visualization,
            model_name="openai:gpt-5-mini",
        ),
    ],
)

Handoff, notebook, viz, and sandbox factories accept model_name and api_key. Without a model, a capability inherits the session model and its API key. With an explicit model, it uses that provider’s configured credentials unless you pass api_key to the factory. An explicit key takes precedence in either case.

Note

SQLSaberOptions.tool_overrides and ModelOverides have been removed. Move each model and API key into its capability factory as shown above. CLI users configure saved plugin models with saber plugins set NAME model PROVIDER:MODEL.

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

Allow 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 and DELETE requiring a WHERE) and limited DDL. Destructive operations such as DROP and TRUNCATE remain blocked. See Running queries for the full policy.

Inject 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. You own its lifecycle. If you omit it, the session creates and closes one automatically.

See Knowledge base for managing entries.

Persist 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 saved connection 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 saved connection is no longer available. All resume errors inherit from ThreadResumeError.

See Conversation threads for CLI thread commands.