Skip to main content
Source: 03_streaming.py · 03-streaming.ts
Use agent.stream(input) to receive lifecycle events as the agent runs. After iterating, call .result() for the final RunResult.

​
Iterate events

import sys
from opperai.agent import Agent, tool

agent = Agent(name="weather", instructions="...", tools=[get_weather])

stream = agent.stream("What's the weather in Paris and Tokyo?")

async for event in stream:
    match event.type:
        case "iteration_start":
            print(f"\n[Iteration {event.iteration}]")
        case "text_delta":
            sys.stdout.write(event.text)
            sys.stdout.flush()
        case "tool_start":
            print(f"-> {event.name}({event.input})")
        case "tool_end":
            print(f"<- {event.name} -> {event.output} ({event.duration_ms:.0f}ms)")

result = await stream.result()
print("\nFinal:", result.output)

​
Event types

EventFired when
iteration_startA new think→act cycle begins.
text_deltaThe model emits a chunk of final text.
tool_start / tool_endA tool is about to run / has completed.
resultThe final RunResult (also returned by .result()).
The full set is exported as AgentStreamEvent in TypeScript and lives in opperai.agent._stream in Python.