Skip to main content

​
Streaming

cascadeflow supports streaming responses via stream_events() for custom UI integration and run_streaming() for terminal output with visual feedback.

​
stream_events()

Returns an async iterator of StreamEvent objects. 
async for event in agent.stream_events("Explain TypeScript"):
    if event.type == StreamEventType.CHUNK:
        print(event.content, end="", flush=True)
    elif event.type == StreamEventType.COMPLETE:
        print(f"\nDone — model: {event.data.get('model')}")

​
StreamEvent

FieldTypeDescription
typeStreamEventTypeEvent type
contentstrContent chunk (for CHUNK events)
datadict[str, Any]Event metadata (model, phase, strategy, etc.)

​
StreamEventType

ValueDescription
STARTStream started
CHUNKContent chunk received
TOOL_CALLTool call detected
COMPLETEStream complete
ERRORError occurred

​
run_streaming()

Higher-level method with built-in visual feedback (pulsing dot indicator). 
result = await agent.run_streaming(
    "Explain quantum mechanics",
    enable_visual=True,
    max_tokens=500,
)
print(f"\nCost: ${result.total_cost:.6f}")

​
Tool Streaming

When tools are involved, cascadeflow uses a tool-aware streaming manager: 
async for event in agent.stream_events(
    "Search and summarize",
    tools=tools,
    tool_executor=executor,
    max_steps=5,
):
    if event.type == StreamEventType.TOOL_CALL:
        print(f"Calling tool: {event.data['tool_name']}")
    elif event.type == StreamEventType.CHUNK:
        print(event.content, end="")

​
ToolStreamEvent

Tool-specific streaming event with additional fields.
FieldTypeDescription
typeToolStreamEventTypeTool event type
tool_calldict[str, Any]Tool call details
tool_resultAnyTool execution result
contentstrContent chunk