Results & Streaming

saber.query() returns a SQLSaberResult. It subclasses str, so it is the agent’s text answer, while also exposing 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 (incl. tool calls / 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.

Controlling run usage

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.

Retrieving complete query results

Complete rows are deliberately absent from model history. Use the descriptors on the result and retrieve bytes through the configured store:

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 means every row returned by SQLsaber remains retrievable through the store;
• UI rendering policy may show a bounded terminal or HTML table; and
• model projection is a deterministic, byte-bounded preview and must not be used for whole-dataset statistics when marked truncated.

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

Analyzing application-owned inputs

Install sqlsaber-notebook and provide a WorkspaceInputResolver when the managed agent should analyze private images, arrays, JSON, or other application-owned files. 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


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),
)
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; and 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.

Persisting notebooks and plots

Install sqlsaber-notebook and configure an artifact store to retain and later retrieve 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/object storage.

from sqlsaber import FilesystemArtifactStore, SQLSaber, SQLSaberOptions

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

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/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.

Inspecting the 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 (text chunks, tool calls, etc.) as they arrive, 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.

Multi-turn conversations

Run follow-up queries on the same SQLSaber instance. Each completed query updates that instance’s conversation history automatically.

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.