Results and streaming

saber.query() returns a SQLSaberResult. It subclasses str, so it is the agent’s text answer and also exposes the full execution details. A SQLSaber instance stores the completed history from each successful query, so a later query on the same instance continues the conversation.

result = await saber.query("How many orders shipped last week?")

print(result)            # the answer, as a string
print(result.usage)      # token usage for this run
print(result.new_messages)  # messages from this run, including tool calls and SQL

Result properties

Property	Description
(the value)	The result is a str containing the agent’s final text answer.
.text	The same final answer as a plain str.
.usage	Aggregate RunUsage statistics for the run, or None when the provider returns no usage data.
.new_messages	The pydantic_ai ModelMessage values created by this query, including model requests, responses, and tool calls.
.messages	Compatibility alias for .new_messages.
.all_messages	The complete pydantic_ai message history after this query, including earlier turns on this instance.
.request_usages	A list of pydantic_ai RequestUsage values in model-request order. Use it to inspect usage for each request in a multi-step run.
.final_context_tokens	The input-token count for the final model request. It measures that request’s context, not aggregate run usage, and is 0 when no value is available.
.query_results	Durable complete-query-result descriptors created during this run.
.artifacts	Durable artifact references published by capabilities during this run.

Run usage limits

Pass Pydantic AI’s UsageLimits to query() when a long-horizon task needs a larger request budget or when the embedding application needs token or tool-call limits:

from pydantic_ai.usage import UsageLimits

result = await saber.query(
    "Investigate the revenue anomaly thoroughly",
    usage_limits=UsageLimits(request_limit=200),
)

When usage_limits is omitted, SQLsaber preserves Pydantic AI’s safe default (request_limit=50). Set a field to None only when the embedding application explicitly intends to disable that limit, for example UsageLimits(request_limit=None).

The managed notebook analyst always shares the parent’s live usage counter, so its model requests are included exactly once in result.usage. When the caller explicitly passes usage_limits, the analyst inherits that budget. When the caller omits usage_limits, the analyst itself is uncapped (request_limit=None) rather than silently acquiring Pydantic AI’s per-agent default. The parent SQLsaber run still retains its safe default.

Complete query results

Complete rows are deliberately absent from model history. Use the descriptors on the result and retrieve bytes through the configured store. On an agent you own, pass the same store into SqlTools and follow Build an agent that keeps complete SQL results.

result = await saber.query(
    "Return daily revenue",
    conversation_id="conversation-123",
    metadata={"tenant_id": "acme"},
)

for descriptor in result.query_results:
    loaded = await saber.get_query_result(
        descriptor,
        conversation_id="conversation-123",
        metadata={"tenant_id": "acme"},
    )
    print(loaded.rows())

These are separate concepts:

• Complete result availability. Every row returned by SQLsaber remains retrievable through the store.
• UI rendering policy. Terminal and HTML views may still show a bounded table.
• Model projection. A deterministic, byte-bounded preview. Do not use it for whole-dataset statistics when it is marked truncated.

Legacy threads containing embedded complete rows remain readable during the compatibility window, but are not automatically imported into durable storage.

Application-owned notebook inputs

Install sqlsaber-notebook and list it in SQLSaberOptions.capabilities when the managed agent should analyze private images, arrays, JSON, or other application-owned files. The CLI lists installed plugins for you. The model sees only opaque attachment_refs. Your resolver converts authorized references into trusted WorkspaceFile bytes:

from collections.abc import Sequence

from sqlsaber import (
    SQLSaber,
    SQLSaberOptions,
    WorkspaceResolutionContext,
)
from sqlsaber_notebook import WorkspaceFile
from sqlsaber_notebook.capability import capability as notebook


class AttachmentResolver:
    def __init__(self, repository):
        self.repository = repository

    async def resolve(
        self,
        refs: Sequence[str],
        *,
        context: WorkspaceResolutionContext,
    ) -> Sequence[WorkspaceFile]:
        # Query by both reference and the current identity. Missing and
        # unauthorized values must be indistinguishable to the caller.
        records = await self.repository.load_authorized(
            refs,
            tenant_id=context.metadata.get("tenant_id"),
            user_id=context.metadata.get("user_id"),
            conversation_id=context.conversation_id,
        )
        return [
            WorkspaceFile(
                name=record.safe_name,
                data=record.data,
                media_type=record.media_type,
                provenance={
                    "attachment_id": record.id,
                    "sha256": record.sha256,
                },
            )
            for record in records
        ]


options = SQLSaberOptions(
    database="sqlite:///analytics.db",
    workspace_input_resolver=AttachmentResolver(repository),
    capabilities=[notebook],
)
async with SQLSaber(options=options) as saber:
    await saber.query(
        "Analyze the selected measurements",
        conversation_id="conversation-123",
        metadata={"tenant_id": "acme", "user_id": "user-456"},
    )

The host, not SQLsaber or the notebook package, owns reference issuance, retrieval, authorization, expiration, and history scoping. Never treat possession of a reference as authorization, and never interpret model-provided values as local paths, URLs, bucket names, or object keys. Raise WorkspaceInputUnavailable with the same bounded message for missing and unauthorized references. SQLsaber hides unexpected resolver exceptions so storage details do not enter model history.

Resolver output is validated before any notebook starts. Names must be visible, single-component filenames. manifest.json is reserved. Bytes must be immutable. Duplicate names or collisions with SQL result names fail the request. The managed workspace allows at most 50 files, 100 MiB per file, and 250 MiB total across SQL results and resolved inputs. Filenames are capped at 255 UTF-8 bytes, and the staged metadata manifest has a separate 1 MiB limit. Explicit SQL result order is preserved, followed by the resolver’s output order. MIME type and string provenance are written to manifest.json.

With no resolver, attachment_refs is omitted from the model-visible tool schema. The existing files selector still means only complete execute_sql result keys. Resolved images are staged as notebook files. They are not automatically sent as initial multimodal content to the child model. The analyst can load and display an image, after which existing bounded PNG snapshot behavior applies.

Notebook and plot artifacts

Pass the notebook factory in SQLSaberOptions.capabilities. Inject an artifact_store to retain executed notebooks, plots, and generated files. Embedded SQLsaber does not persist artifacts unless the application injects a store. SQLsaber includes filesystem and in-memory stores. Cloud applications can implement the same ArtifactStore protocol for private database or object storage.

from sqlsaber import FilesystemArtifactStore, SQLSaber, SQLSaberOptions
from sqlsaber_notebook.capability import capability as notebook

store = FilesystemArtifactStore("/private/sqlsaber-artifacts")
options = SQLSaberOptions(
    database="sqlite:///analytics.db",
    artifact_store=store,
    capabilities=[notebook],
)

async with SQLSaber(options=options) as saber:
    result = await saber.query(
        "Analyze revenue anomalies and plot them",
        conversation_id="conversation-123",
        metadata={"tenant_id": "acme", "user_id": "user-456"},
    )

    for descriptor in result.artifacts:
        loaded = await saber.get_artifact(
            descriptor,
            conversation_id="conversation-123",
            metadata={"tenant_id": "acme", "user_id": "user-456"},
        )
        print(descriptor.kind, descriptor.name, descriptor.uri, len(loaded.data))

ArtifactStore.publish() receives an ArtifactBundle and current ArtifactContext. get() retrieves one artifact by opaque ID. A cloud store must authorize retrieval from the current context metadata, return ArtifactUnavailable for both missing and unauthorized IDs, and return stable private object keys or internal URIs rather than expiring signed URLs. Generate signed download URLs only in the host application’s serving layer.

The store is application-owned and SQLsaber never closes or garbage-collects it. Keep object storage private because executed notebooks may contain query results. The SQLsaber CLI separately configures private local storage under its user-data directory and applies thread-aware retention.

With artifact_failure_mode="required", an upload failure fails analyze_data. Use "best_effort" to preserve the analyst’s answer and expose an artifact_error in tool metadata instead.

The same application-owned store also supports direct sqlsaber-notebook embedding. Analysis remains storage-independent. Publish its completed result in a second explicit operation:

from sqlsaber import ArtifactContext
from sqlsaber_notebook import Workspace, analyze, publish_analysis

analysis = await analyze(
    "Plot monthly revenue",
    Workspace.from_files([("revenue.csv", revenue_bytes)]),
    model="anthropic:claude-sonnet-4-6",
    model_provider="anthropic",
    collect_files=True,
)
publication = await publish_analysis(
    analysis,
    store=store,
    context=ArtifactContext(
        conversation_id="conversation-123",
        metadata={"tenant_id": "acme", "user_id": "user-456"},
    ),
)

Managed SQLsaber calls this same publication operation. The standalone sqlsaber-notebook --output analysis.ipynb command is different: it writes the explicit notebook and sibling artifact directory and never silently uses SQLsaber’s user-data store.

Generated SQL

The SQL the agent ran lives in the run’s messages as tool calls. Message values and streaming event values use pydantic_ai types, not framework-neutral SDK transcript or stream types. Iterate over message parts to find tool calls:

from pydantic_ai.messages import ModelResponse, ToolCallPart

result = await saber.query("Top 5 customers by revenue")

for message in result.new_messages:
    if isinstance(message, ModelResponse):
        for part in message.parts:
            if isinstance(part, ToolCallPart):
                print(part.tool_name, part.args)

Streaming events

Pass an event_stream_handler to react to events as they arrive, such as text chunks and tool calls, rather than waiting for the final answer. The handler is an async function that receives the run context and an async iterable of events:

from collections.abc import AsyncIterable
from typing import Any

from pydantic_ai import RunContext
from pydantic_ai.messages import AgentStreamEvent


async def on_event(
    ctx: RunContext[Any],
    events: AsyncIterable[AgentStreamEvent],
) -> None:
    async for event in events:
        print(event)
        # execute_sql FunctionToolResultEvent parts expose the bounded projection
        # in part.content and the descriptor in part.metadata["query_result"].


result = await saber.query(
    "Show me revenue by month",
    event_stream_handler=on_event,
)

Note

The event and context types come from pydantic-ai, which powers SQLsaber’s agent. Refer to its docs for the full event taxonomy.

A web backend can detect part.metadata["query_result"], emit a query-result-created SSE event containing the ID/columns/row count, and serve an authorized paginated or download endpoint separately. SQLsaber never creates a public URL. A typical production store writes descriptor and tenant ownership columns to PostgreSQL and stores payload bytes in a private bytea column (or an object key for private blob storage). Its get implementation should query by both result_id and the current tenant_id from QueryResultContext.metadata, then raise QueryResultUnavailable for both missing and unauthorized rows. Apply application retention, encryption, audit, and deletion policies there. Filesystem storage is not suitable for horizontally scaled web deployments.

Steer a running query

query() is one pydantic-ai run. steer() appends a user message to that same run. pydantic-ai delivers it before the next model request. SQLSaber does not keep a second queue. pending_steers is read from the run’s live queue.

import asyncio
from sqlsaber import SQLSaber, SQLSaberOptions, SteerIdleError

async def main() -> None:
    async with SQLSaber(options=SQLSaberOptions(database="analytics")) as saber:
        run = asyncio.create_task(saber.query("Revenue by region for 2025"))
        await asyncio.sleep(0)

        enqueue_id = saber.steer("Only US customers, and use last quarter.")
        assert saber.pending_steers == ("Only US customers, and use last quarter.",)

        dropped = saber.clear_steers()
        assert dropped == ["Only US customers, and use last quarter."]
        assert saber.pending_steers == ()

        saber.steer("Only US customers.")
        result = await run

        assert any("Only US customers." in str(m) for m in saber.messages)
        assert saber.pending_steers == ()

        try:
            saber.steer("too late")
        except SteerIdleError as exc:
            assert str(exc) == "No query is running."

steer, clear_steers, and pending_steers are synchronous and must run on the event loop that is running query(). From another thread, marshal with loop.call_soon_threadsafe(saber.steer, text). Calling from the wrong thread raises RuntimeError instead of racing the drain.

steer raises SteerIdleError when no run can take the message, ValueError for an empty message, and SQLSaberClosedError after close(). Two steer calls queue two items. clear_steers does not abort the run.

A steer sent before the first model request lands in that first request. A steer sent while the model or a tool is busy lands in the next request. A steer sent as the agent was about to stop starts one more model request. The current token stream is never cut. Abort discards the whole turn, delivered and pending steers included. pending_steers is () afterwards.

Follow-up queries

Run a later query on the same SQLSaber instance after the previous query() returns. Each completed query updates that instance’s conversation history automatically. This starts a new run. Use steer() to redirect the run that is already in flight.

async with SQLSaber(options=options) as saber:
    first = await saber.query("How many active customers do we have?")
    follow_up = await saber.query("Now break that down by country")
    print(follow_up.text)

Use result.all_messages to inspect the complete history. query() still accepts message_history as an advanced compatibility override when an application must supply an explicit pydantic_ai history.