stream_text

API reference for streaming text generation

stream_text() starts a streaming model/tool loop and returns a TextStreamResult immediately.

It is not an async function. Streaming begins when you iterate over result.text_stream.

Import

from ai_query import stream_text

Signature

def stream_text(
    *,
    model: LanguageModel,
    prompt: str | None = None,
    system: str | None = None,
    messages: list[Message] | list[dict[str, Any]] | None = None,
    tools: ToolSet | None = None,
    stop_when: StopCondition | list[StopCondition] | None = None,
    on_step_start: OnStepStart | None = None,
    on_step_finish: OnStepFinish | None = None,
    on_reasoning_event: OnReasoningEvent | None = None,
    retry: RetryPolicy | None = None,
    on_retry: OnRetry | None = None,
    provider_options: ProviderOptions | None = None,
    reasoning: ReasoningConfig | None = None,
    signal: AbortSignal | None = None,
    **kwargs: Any,
) -> TextStreamResult

Parameters

model · LanguageModel · required

Language model instance.

prompt · str | None

Convenience user prompt. Ignored when messages is provided.

system · str | None

Optional system prompt. Only used when messages is not provided.

messages · list[Message] | list[dict[str, Any]] | None

Explicit message list. When provided, prompt and system are ignored.

tools · ToolSet | None

Mapping of tool name to Tool.

stop_when · StopCondition | list[StopCondition] | None

Stop condition or list of stop conditions. Defaults to step_count_is(1).

on_step_start · OnStepStart | None

Called before each provider request. May return StepControl.

on_step_finish · OnStepFinish | None

Called after each step completes.

on_reasoning_event · OnReasoningEvent | None

Called for provider reasoning/thinking events. These are not mixed into text_stream.

retry · RetryPolicy | None

Provider-call retry policy. By default, retries are conservative: transient HTTP statuses (408, 409, 425, 429, 500, 502, 503, 504) and network/timeout failures are retried; clear client errors such as 400, 401, and 403 are not. Streaming retries only happen before the provider has emitted any output for that attempt.

on_retry · OnRetry | None

Called before a retry attempt with the step number, attempt, delay, and sanitized error string.

provider_options · ProviderOptions | None

Provider-specific options keyed by provider name.

reasoning · ReasoningConfig | None

Provider-agnostic reasoning controls.

signal · AbortSignal | None

Abort signal used to cancel the streaming loop, including in-flight provider stream reads, retry sleeps, and async tool calls.

**kwargs · Any

Additional provider-specific keyword arguments passed to the provider stream call.

Returns

TextStreamResult · TextStreamResult

Lazy streaming result object.

Properties:

event_stream · AsyncIterator[TextStreamEvent]

Async iterator yielding typed events in generation order. Event types are step.started, text.delta, reasoning.summary, reasoning.delta, reasoning.signature, reasoning.state, tool_call.started, tool_call.delta, tool_call.ready, tool_execution.started, tool_execution.finished, tool_result, step.finished, and stream.finished.

text_stream · AsyncIterator[str]

Text-only projection of event_stream.

text · Awaitable[str]

Complete accumulated text. Await after or during stream consumption.

usage · Awaitable[Usage | None]

Usage after the stream completes.

finish_reason · Awaitable[str | None]

Finish reason after the stream completes.

steps · Awaitable[list[StepResult]]

Step history after the stream completes.

TextStreamEvent

TextStreamEvent is a discriminated union. Branch on event.type before reading event-specific fields.

Type	Event class	Fields
step.started	StreamStepStartedEvent	step_number, messages, tools
text.delta	TextDeltaEvent	text, step_number
reasoning.*	StreamReasoningEvent	event, step_number
tool_call.started	ToolCallStartedEvent	step_number, index, tool_call_id, name
tool_call.delta	ToolCallDeltaEvent	step_number, index, tool_call_id, name_delta, arguments_delta
tool_call.ready	ToolCallReadyEvent	step_number, index, tool_call
tool_execution.started	ToolExecutionStartedEvent	step_number, index, tool_call
tool_execution.finished	ToolExecutionFinishedEvent	step_number, index, tool_call, tool_result, duration, error, aborted
tool_result	ToolResultEvent	step_number, index, tool_call, tool_result
step.finished	StreamStepFinishedEvent	step_number, step, text, usage, steps
stream.finished	StreamFinishedEvent	text, finish_reason, usage, steps

The reasoning.* discriminator preserves the underlying ReasoningEvent.kind: summary, delta, signature, or state. Treat reasoning.summary and reasoning.delta as streamed reasoning output; they are intentionally separate from text.delta and are not included in text_stream. Step history fields are snapshots of the completed steps at the time that event was emitted.

Tool calls receive a stable provider-order index. Ready and result events are emitted in provider order. Parallel execution finish events are emitted in actual completion order. tool_execution.finished captures the direct execution result; tool_result contains the final result after after_tool_call hooks and matches the result stored in StepResult.tool_results.

Blocked and unknown tools emit tool_call.ready and an error tool_result, but do not emit execution start or finish events because no tool function ran.

Every completed tool call emits tool_call.started, at least one tool_call.delta, then tool_call.ready. Concatenate name_delta and arguments_delta by (step_number, index); IDs may be unavailable on the first fragment and appear on a later delta. tool_call.ready is the normalization boundary and contains the parsed ToolCall.

Provider coverage:

Provider family	Fragment source	Notes
OpenAI-compatible	upstream fragments	Streams OpenAI-style delta.tool_calls fragments as they arrive.
Anthropic	upstream fragments	Streams tool_use block starts and input_json_delta fragments.
Bedrock	upstream fragments	Streams toolUse block starts and input deltas.
Google Gemini	normalized from completed calls	Gemini generally returns completed functionCall parts, so ai-query synthesizes start/delta immediately before tool_call.ready.

Always handle tool_call.ready as the authoritative parsed call. Treat start/delta events as progressive updates when the provider exposes fragments, or as normalized one-shot call previews when it does not.

Raises

Errors occur while consuming the stream, not when calling stream_text().

ValueError · Exception

Raised when neither prompt nor messages is provided.

AbortError · Exception

Raised while consuming if signal is aborted.

Examples

Typed event stream

result = stream_text(
    model=model,
    prompt="Solve this problem step by step.",
    reasoning={"effort": "high"},
)

async for event in result.event_stream:
    if event.type == "text.delta":
        print(event.text, end="", flush=True)
    elif event.type.startswith("reasoning.") and event.event.text:
        print(f"{event.event.kind}: {event.event.text}")
    elif event.type == "step.finished":
        record_step(event.step)

The event stream is replayable and shares its source with text_stream. Reading both views does not make a second provider request.

Basic stream

result = stream_text(
    model=openai("gpt-4o"),
    prompt="Write a haiku about Python.",
)

async for chunk in result.text_stream:
    print(chunk, end="", flush=True)

Read final values

result = stream_text(model=model, prompt="Explain recursion.")

async for chunk in result.text_stream:
    print(chunk, end="")

text = await result.text
usage = await result.usage
steps = await result.steps

Reasoning events

async def on_reasoning(event):
    if event.text:
        print(f"thinking: {event.text}")


result = stream_text(
    model=model,
    prompt="Solve this problem step by step.",
    reasoning={"effort": "high"},
    on_reasoning_event=on_reasoning,
)

on_reasoning_event remains supported for compatibility. New integrations that need text, reasoning, and lifecycle ordering should consume event_stream.

Streaming with tools

result = stream_text(
    model=model,
    prompt="Search docs and summarize installation.",
    tools={"search_docs": search_docs},
    stop_when=step_count_is(5),
)

async for chunk in result.text_stream:
    print(chunk, end="", flush=True)

Show retry status

from ai_query import RetryPolicy


async def on_retry(event):
    await publish_status({"phase": "retrying", "attempt": event.attempt})


result = stream_text(
    model=model,
    prompt="Generate the report.",
    retry=RetryPolicy(max_attempts=3),
    on_retry=on_retry,
)

See Also

• generate_text
• Results
• Thinking
• Streaming