> ## Documentation Index
> Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming

> Stream AI responses in real-time

The `stream_text` function enables real-time streaming of AI responses. This provides a better user experience by showing text as it's generated rather than waiting for the complete response.

`stream_text` exposes a unified typed `event_stream` for text, provider thinking,
and lifecycle events. Its `text_stream` property remains a text-only projection,
and `on_reasoning_event` remains available for callback-based integrations.

## Basic Streaming

```python theme={null}
import asyncio
from ai_query import stream_text
from ai_query.providers import google

async def main():
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt="Write a short story about a robot."
    )

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

    print()  # New line at the end

asyncio.run(main())
```

<Note>
  `stream_text` returns immediately with a [`TextStreamResult`](/reference/types/results) object. The provider request starts when you iterate over `event_stream` or `text_stream`, or await a final result property.
</Note>

## Getting Usage Statistics

Token usage is available after streaming completes:

```python theme={null}
result = stream_text(
    model=google("gemini-2.0-flash"),
    prompt="Explain machine learning."
)

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

# Await usage after streaming
usage = await result.usage
print(f"\nTokens: {usage.total_tokens}")
```

## Streaming with Tools

Combine streaming with tool calling for powerful agentic workflows:

```python theme={null}
from ai_query import stream_text
from ai_query.providers import google, tool, Field

@tool(description="Search Wikipedia for information")
async def search_wikipedia(query: str = Field(description="Search query")) -> str:
    # Simulated search
    return f"Wikipedia results for '{query}': ..."

async def main():
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt="Search for information about Python programming and summarize it.",
        tools={"search_wikipedia": search_wikipedia}
    )

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

asyncio.run(main())
```

When tools are provided, the streaming handles the full execution loop:

1. The AI decides to call a tool
2. The tool executes
3. The AI receives the result
4. The AI generates the final response (streamed)

## Parameters

| Parameter            | Type                                                | Description                                                    |
| -------------------- | --------------------------------------------------- | -------------------------------------------------------------- |
| `model`              | `LanguageModel`                                     | The AI model to use                                            |
| `prompt`             | `str`                                               | The user prompt                                                |
| `messages`           | [`list[Message]`](/reference/types/message)         | Full conversation history                                      |
| `system`             | `str`                                               | System prompt                                                  |
| `tools`              | [`dict[str, Tool]`](/reference/types/tool)          | Available tools                                                |
| `stop_when`          | [`StopCondition`](/reference/types/stop-conditions) | Stop condition for tool loops                                  |
| `on_step_start`      | `Callable`                                          | Step start callback                                            |
| `on_step_finish`     | `Callable`                                          | Step finish callback                                           |
| `on_reasoning_event` | `Callable`                                          | Callback for provider thinking output during streaming         |
| `reasoning`          | `ReasoningConfig`                                   | Normalized reasoning-depth controls like `effort` and `budget` |
| `provider_options`   | `dict`                                              | Provider-specific options                                      |

## Typed Events

Consume `event_stream` when one ordered channel needs to drive a UI, transport,
or log:

```python theme={null}
result = stream_text(model=model, prompt="Explain binary search.")

async for event in result.event_stream:
    if event.type == "text.delta":
        render_text(event.text)
    elif event.type.startswith("reasoning."):
        render_reasoning(event.event)
    elif event.type == "step.finished":
        record_step(event.step)
```

The stream emits `step.started` and `step.finished` around each model step, then
one `stream.finished` event. Both stream views are replayable and share one
provider request, including when consumed concurrently.

When a completed tool call is available, the same stream emits
`tool_call.ready`, `tool_execution.started`, `tool_execution.finished`, and
`tool_result`. Parallel finish events arrive as tools complete; normalized result
events remain in provider order.

OpenAI-compatible, Anthropic, and Bedrock streams also emit `tool_call.started`
and `tool_call.delta` while function names and JSON arguments are being assembled.
Use the step number and provider index for correlation until a tool call ID is
available. Providers that only return completed calls, such as Gemini
`functionCall` parts, still get normalized start/delta events immediately before
`tool_call.ready`.

## Thinking Events

Reasoning is emitted as `reasoning.summary`, `reasoning.delta`,
`reasoning.signature`, or `reasoning.state`. Summary and delta events are both
streamed reasoning output, but neither is included in `text_stream`.
Callback-based integrations can continue using `on_reasoning_event`:

```python theme={null}
from ai_query import stream_text
from ai_query.providers import google

def on_reasoning(event):
    print(f"\n[{event.kind}]", event.text or event.data)

result = stream_text(
    model=google("gemini-2.5-pro"),
    prompt="Explain binary search.",
    provider_options={
        "google": {
            "thinking_config": {
                "thinking_budget": 1024,
                "include_thoughts": True,
            }
        }
    },
    on_reasoning_event=on_reasoning,
)

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

See [`Thinking`](/core/thinking) for provider differences and guidance.

## TextStreamResult Object

The `stream_text` function returns a [`TextStreamResult`](/reference/types/results):

```python theme={null}
result = stream_text(model=google("gemini-2.0-flash"), prompt="Hello")

# Async iterator for text chunks
async for chunk in result.text_stream:
    print(chunk, end="")

# Await to get final values
usage = await result.usage        # Token usage
text = await result.text          # Complete text (after streaming)
tool_calls = await result.tool_calls  # Any tool calls made
```

## Collecting Full Text

If you need the complete text after streaming:

```python theme={null}
result = stream_text(
    model=google("gemini-2.0-flash"),
    prompt="Write a poem."
)

# Stream to console
async for chunk in result.text_stream:
    print(chunk, end="")

# Get full text
full_text = await result.text
print(f"\n\nFull response ({len(full_text)} chars)")
```

## Real-time UI Updates

Stream directly to a user interface:

```python theme={null}
async def stream_to_ui(prompt: str, update_callback):
    """Stream AI response to a UI callback."""
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt=prompt
    )

    buffer = ""
    async for chunk in result.text_stream:
        buffer += chunk
        await update_callback(buffer)

    return buffer
```

## Next Steps

<CardGroup cols={2}>
  <Card title="Tools" icon="wrench" href="/core/tools">
    Add function calling to your streams
  </Card>

  <Card title="Agents" icon="robot" href="/core/agents">
    Build agents with streaming output
  </Card>
</CardGroup>