How to Configure Streaming Responses with `stream` and `stream_events` in Agno

To enable streaming in Agno, set stream=True on your workflow's run() method to receive a generator of final outputs, and add stream_events=True to emit intermediate events like step starts and tool calls during execution.

The Agno framework provides granular control over how workflow results are delivered through two complementary boolean flags. These parameters determine whether you receive a single final object or a real-time iterator of events, making it possible to build responsive applications that display progress as it happens.

Understanding stream vs. stream_events

Agno's Workflow class supports two distinct streaming modes defined in libs/agno/agno/workflow/workflow.py (lines 200-206):

• stream: When set to True, the run() method returns a generator yielding WorkflowRunOutputEvent objects containing the final workflow output. When False or None, it returns a single WorkflowRunOutput object.
• stream_events: When enabled alongside stream=True, the iterator yields intermediate events such as WorkflowStartedEvent, StepOutputEvent, and WorkflowCompletedEvent, allowing you to track execution progress in real time.

If stream is False, the source code explicitly forces stream_events to False (see lines 989-999 in workflow.py), ensuring no event overhead occurs during standard synchronous execution.

How Stream Configuration Works in the Source Code

The streaming logic is implemented in the run() method starting at line 3893 of libs/agno/agno/workflow/workflow.py.

Execution Path Selection

The workflow determines which execution path to take based on the stream parameter:

1. Streaming path: If stream=True, the code calls Workflow._execute_stream() (referenced at lines 558-566), which yields events as they occur.
2. Standard path: If stream=False, the code executes _execute() and returns a single WorkflowRunOutput object.

Configuration Hierarchy

Agno respects a cascading configuration system:

1. Instance defaults: Set stream and stream_events during Workflow initialization to establish defaults for all runs.
2. Per-run overrides: Pass explicit arguments to run() or arun() to override instance defaults for specific executions.
3. Helper enforcement: Utilities like print_response_stream (defined in libs/agno/agno/utils/print_response/workflow.py, lines 95-104) internally force stream_events=True to guarantee event availability for live console rendering.

Practical Implementation Examples

Basic Streaming of Final Output

Use stream=True to iterate over the final result as it becomes available. This mode yields only the completed workflow output without intermediate step details.

from agno import Workflow

wf = Workflow(name="demo", steps=my_steps)

# Returns a generator yielding WorkflowRunOutputEvent objects

for event in wf.run(input="hello", stream=True):
    print(event.output)  # Final workflow result

Full Event Streaming with Intermediate Steps

Enable both flags to receive a rich stream of events including step transitions and progress updates. This is essential for building live dashboards or progress indicators.

from agno import Workflow, WorkflowStartedEvent, StepOutputEvent

wf = Workflow(name="demo", steps=my_steps)

for ev in wf.run(input="hello", stream=True, stream_events=True):
    if isinstance(ev, WorkflowStartedEvent):
        print("🚀 Workflow started")
    elif isinstance(ev, StepOutputEvent):
        print(f"🔹 Step {ev.step_index}: {ev.output}")
    else:
        # WorkflowRunOutputEvent indicates completion

        print("✅ Final output:", ev.output)

Using the Built-in Print Helper

The print_response_stream utility automates event streaming and provides a polished console interface using Rich. According to the source code in libs/agno/agno/utils/print_response/workflow.py, this helper automatically enables stream_events=True internally.

from agno import Workflow
from agno.utils.print_response.workflow import print_response_stream

wf = Workflow(name="demo", steps=my_steps)

# Automatically handles stream=True and stream_events=True

print_response_stream(
    workflow=wf,
    input="Summarize the last 5 chat messages"
)

Async Streaming with arun

For asynchronous applications, use arun() with identical streaming parameters. The async implementation follows the same event-yielding pattern as the synchronous version.

import asyncio
from agno import Workflow

async def main():
    wf = Workflow(name="demo", steps=my_steps)
    
    async for event in wf.arun(
        input="process data",
        stream=True,
        stream_events=True
    ):
        # Handle events identically to sync version

        if hasattr(event, 'step_index'):
            print(f"Step {event.step_index} complete")

asyncio.run(main())

Setting Workflow-Level Defaults

Configure default streaming behavior at the workflow level to avoid repeating parameters on every run() call. These defaults can still be overridden per execution.

wf = Workflow(
    name="demo",
    steps=my_steps,
    stream=True,           # Default to streaming

    stream_events=True,    # Default to full event stream

)

# Uses defaults (streaming enabled)

wf.run(input="test")

# Override defaults for non-streaming execution

wf.run(input="test", stream=False)

Summary

• stream=True converts the run() return value from a single object to a generator of WorkflowRunOutputEvent objects containing final results.
• stream_events=True (requires stream=True) expands the generator to include intermediate events like WorkflowStartedEvent and StepOutputEvent for real-time progress tracking.
• The source implementation in libs/agno/agno/workflow/workflow.py forces stream_events=False when stream=False to prevent unnecessary overhead.
• Use print_response_stream from libs/agno/agno/utils/print_response/workflow.py for turnkey console-based streaming UIs that automatically handle event configuration.

Frequently Asked Questions

What happens if I set stream_events=True but stream=False?

According to the source code in libs/agno/agno/workflow/workflow.py (lines 989-999), the run() method automatically forces stream_events=False when stream is disabled. You must set stream=True to receive any events, including intermediate ones.

Can I change streaming settings after creating a Workflow instance?

Yes. While you can set defaults during initialization in the Workflow dataclass (lines 200-206), the run() and arun() methods accept explicit stream and stream_events parameters that override instance defaults for that specific execution.

What types of events does stream_events=True emit?

When enabled, the generator yields various event types including WorkflowStartedEvent, StepOutputEvent, WorkflowAgentFinishedEvent, and finally WorkflowRunOutputEvent. The exact composition depends on your workflow's step configuration and whether agents are involved in individual steps.

Is there a performance penalty for enabling stream_events?

Yes, there is minimal overhead associated with event object instantiation and generator yielding. However, for most use cases, the latency is negligible compared to the actual execution time of workflow steps. If maximum throughput is required and you don't need real-time feedback, use stream=False.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}