Streaming responses

agent() defaults to stream=True. Each token the LLM produces goes to connected clients as it arrives. The final assembled string is what Flux records in the event log.

How token streaming works

With stream=True, agent_loop calls the provider’s async generator interface instead of its blocking call. As each token arrives, the loop calls progress() with {"token": <text>}. Clients connected to the /run/stream endpoint receive these as Server-Sent Events.

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent

@workflow
async def summarizer(ctx: ExecutionContext):
    assistant = await agent(
        "You are a concise summarizer.",
        model="ollama/llama3.2",
        stream=True,          # default — shown explicitly for clarity
    )
    return await assistant(ctx.input["text"])

Start a streaming session with curl:

curl -N -X POST http://localhost:8000/workflows/summarizer/run/stream \
  -H "Content-Type: application/json" \
  -d '{"text": "The quick brown fox..."}'

Each SSE frame looks like:

data: {"type": "TASK_PROGRESS", "value": {"token": "The"}}
data: {"type": "TASK_PROGRESS", "value": {"token": " document"}}
data: {"type": "TASK_PROGRESS", "value": {"token": " describes"}}
...
data: {"type": "WORKFLOW_COMPLETED", "value": "The document describes..."}

stream=True on the agent is the only setup needed. Token-to-SSE routing happens inside the agent loop automatically.

What gets checkpointed

progress() events are ephemeral. They reach connected clients in real time but are never written to the database or replayed during workflow recovery.

The complete response is durable. After the generator is exhausted, the assembled text is returned as the task’s output and recorded in the execution event log as a TASK_COMPLETED event. Retries, replays, and workflow resumes all operate on this final value.

When tools are in use alongside streaming, each LLM call gets wrapped in a named task (llm_0, llm_1, …). If a tool triggers a human-approval pause and the workflow later resumes, the LLM is not re-invoked — the checkpointed response is replayed, preserving the same tool calls that existed before the pause.

Provider support

All four providers implement the stream() method on their formatter class. The mechanics differ per provider, but the agent_loop interface is the same from the workflow’s perspective.

All four providers set supports_reasoning_stream = True. When reasoning_effort is set, thinking tokens are also streamed via progress({"type": "reasoning", "text": <chunk>}).

Streaming with tools

When the agent has tools, agent_loop calls call_with_reasoning_stream rather than the basic stream() path. Tool-call events also go through progress():

• {"type": "tool_start", "id": ..., "name": ..., "args": ...} when the LLM requests a tool call
• {"type": "tool_done", "id": ..., "name": ..., "status": "success"|"error"} after the tool returns

Like token events, these are ephemeral. A UI can use them to show which tool is running and whether it succeeded.

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent

@task
async def web_search(query: str) -> str:
    """Search the web and return a summary of results."""
    # implementation omitted
    return f"Results for: {query}"

@workflow
async def research_agent(ctx: ExecutionContext):
    assistant = await agent(
        "You are a research assistant. Use web_search to find current information.",
        model="openai/gpt-4o-mini",
        tools=[web_search],
        stream=True,
    )
    return await assistant(ctx.input["question"])

The SSE stream interleaves token events and tool events in order, so a client has enough to render a live activity feed.

Disabling streaming

Pass stream=False to get the complete response in one blocking call. This can work better when the caller does not need incremental output, or when integrating with systems that buffer responses on receipt.

assistant = await agent(
    "You are a concise summarizer.",
    model="anthropic/claude-sonnet-4-20250514",
    stream=False,
)

The durability behavior is the same either way — the final response is checkpointed as a TASK_COMPLETED event.

Consuming SSE in Python

The /run/stream endpoint uses standard SSE. Here is a minimal example using httpx:

import httpx

url = "http://localhost:8000/workflows/summarizer/run/stream"
payload = {"text": "..."}

with httpx.Client(timeout=None) as client:
    with client.stream("POST", url, json=payload) as response:
        for line in response.iter_lines():
            if line.startswith("data:"):
                import json
                event = json.loads(line[5:].strip())
                if event.get("type") == "TASK_PROGRESS":
                    token = event["value"].get("token", "")
                    print(token, end="", flush=True)

For an example that batches tokens into task events — one frame per batch rather than one per token — see examples/ai/streaming_with_task_events_ollama.py in the source repository.

For workflows that emit progress from ordinary tasks (not agent streaming), see Workflows — Streaming output.