Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 26 additions & 11 deletions lib/crewai/src/crewai/types/streaming.py
Original file line number Diff line number Diff line change
Expand Up @@ -504,13 +504,15 @@ class CrewStreamingOutput(StreamingOutputBase["CrewOutput"]):

Example:
```python
# Single crew
# Single crew — the crew must be constructed with stream=True
crew = Crew(agents=[...], tasks=[...], stream=True)
streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
print(chunk.content, end="", flush=True)
result = streaming.result

# Multiple crews (kickoff_for_each_async)
# Multiple crews (kickoff_for_each_async) — also requires stream=True
crew = Crew(agents=[...], tasks=[...], stream=True)
streaming = await crew.kickoff_for_each_async(
[{"topic": "AI"}, {"topic": "ML"}]
)
Expand Down Expand Up @@ -580,22 +582,35 @@ class FlowStreamingOutput(StreamingOutputBase[Any]):
"""Streaming output wrapper for flow execution.

Provides both sync and async iteration over stream chunks,
with access to the final flow output via the .result property.
with access to the final flow output via the ``.result`` property.

Example:
```python
# Sync usage
streaming = flow.kickoff_streaming()
for chunk in streaming:
print(chunk.content, end="", flush=True)
# Flow-level streaming returns a StreamSession from Flow.kickoff() —
# NOT a FlowStreamingOutput. See
# docs/edge/en/learn/streaming-flow-execution.mdx for the full guide.
flow = MyFlow()
flow.stream = True
streaming = flow.kickoff() # -> StreamSession
for frame in streaming:
print(frame.content, end="", flush=True)
result = streaming.result

# Async usage
streaming = await flow.kickoff_streaming_async()
async for chunk in streaming:
print(chunk.content, end="", flush=True)
# Async variant:
flow = MyFlow()
flow.stream = True
streaming = await flow.kickoff_async() # -> AsyncStreamSession
async for frame in streaming:
print(frame.content, end="", flush=True)
result = streaming.result
```

Note:
Flow-level streaming is exposed to users through
:class:`StreamSession`; configure the Flow with ``stream=True``
before calling ``Flow.kickoff()``. ``FlowStreamingOutput`` is
retained for consumers that build a streaming wrapper directly
from an existing iterator.
Comment thread
Vidit-Ostwal marked this conversation as resolved.
"""

def _set_result(self, result: Any) -> None:
Expand Down
Loading