From 9371b958869ea0073ddabf7889012e7dcbf51a9a Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Fri, 3 Jul 2026 20:49:36 +0800 Subject: [PATCH 1/3] feat(crew): add stream_frames opt-in for full event frame streaming When stream=True and stream_frames=True, kickoff/kickoff_async now stream full StreamFrame events (LLM tokens, tool calls, agent/task lifecycle) via the existing frame pipeline, instead of the default StreamChunk token-only stream. Returns a StreamSession / AsyncStreamSession whose .subscribe(channels=...) allows filtering by channel (llm, tools, messages, flow, lifecycle). Reuses the existing frame infrastructure (create_frame_streaming_state / create_frame_generator / stream_frame_from_event) already used by BaseLLM.stream_events() and Flow.stream_events(); no new event emission points and no new output types. The default chunk path (stream=True, stream_frames=False) is unchanged, so the change is fully backward compatible. --- lib/crewai/src/crewai/crew.py | 104 +++++++++++++++++++++++++++++++++- 1 file changed, 101 insertions(+), 3 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index fda456be40..b5db421f9e 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -117,7 +117,11 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.tools.agent_tools.read_file_tool import ReadFileTool from crewai.tools.base_tool import BaseTool from crewai.types.callback import SerializableCallable -from crewai.types.streaming import CrewStreamingOutput +from crewai.types.streaming import ( + AsyncStreamSession, + CrewStreamingOutput, + StreamSession, +) from crewai.types.usage_metrics import UsageMetrics from crewai.utilities.constants import NOT_SPECIFIED, TRAINING_DATA_FILE from crewai.utilities.crew.models import CrewContext @@ -137,7 +141,10 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.utilities.rpm_controller import RPMController from crewai.utilities.streaming import ( create_async_chunk_generator, + create_async_frame_generator, create_chunk_generator, + create_frame_generator, + create_frame_streaming_state, register_cleanup, signal_end, signal_error, @@ -293,6 +300,16 @@ class Crew(FlowTrackable, BaseModel): default=False, description="Whether to stream output from the crew execution.", ) + stream_frames: bool = Field( + default=False, + description=( + "When True (requires stream=True), kickoff yields StreamFrame " + "events covering the full execution (LLM tokens, tool calls, " + "agent/task lifecycle) instead of StreamChunk token chunks. " + "Returns a StreamSession / AsyncStreamSession whose .subscribe() " + "allows filtering by channel (llm, tools, messages, flow, lifecycle)." + ), + ) max_rpm: int | None = Field( default=None, description=( @@ -968,7 +985,7 @@ def kickoff( inputs: dict[str, Any] | None = None, input_files: dict[str, FileInput] | None = None, from_checkpoint: CheckpointConfig | None = None, - ) -> CrewOutput | CrewStreamingOutput: + ) -> CrewOutput | CrewStreamingOutput | StreamSession[CrewOutput]: """Execute the crew's workflow. Args: @@ -987,6 +1004,8 @@ def kickoff( get_env_context() if self.stream: enable_agent_streaming(self.agents) + if self.stream_frames: + return self._kickoff_frame_stream(inputs, input_files) ctx = StreamingContext() def run_crew() -> None: @@ -1098,7 +1117,7 @@ async def kickoff_async( inputs: dict[str, Any] | None = None, input_files: dict[str, FileInput] | None = None, from_checkpoint: CheckpointConfig | None = None, - ) -> CrewOutput | CrewStreamingOutput: + ) -> CrewOutput | CrewStreamingOutput | AsyncStreamSession[CrewOutput]: """Asynchronous kickoff method to start the crew execution. Args: @@ -1121,6 +1140,8 @@ async def kickoff_async( if self.stream: enable_agent_streaming(self.agents) + if self.stream_frames: + return await self._akickoff_frame_stream(inputs, input_files) ctx = StreamingContext(use_async=True) async def run_crew() -> None: @@ -1147,6 +1168,83 @@ async def run_crew() -> None: return await asyncio.to_thread(self.kickoff, inputs, input_files) + def _kickoff_frame_stream( + self, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + ) -> StreamSession[CrewOutput]: + """Run kickoff and stream full execution frames. + + Yields ``StreamFrame`` events for the entire crew run (LLM tokens, + tool calls, agent/task lifecycle) via the existing frame streaming + pipeline, instead of the default ``StreamChunk`` token-only stream. + Consumers iterate the returned ``StreamSession`` (or filter with + ``.subscribe(channels=...)``) and access the final ``CrewOutput`` via + ``.result`` after exhaustion. Requires ``stream=True``; ``stream_frames`` + selects this path over chunk streaming. + + Args: + inputs: Optional input dictionary for task interpolation. + input_files: Optional dict of named file inputs for the crew. + + Returns: + A ``StreamSession`` yielding ``StreamFrame`` events. + """ + result_holder: list[CrewOutput] = [] + state = create_frame_streaming_state(result_holder, use_async=False) + output_holder: list[StreamSession[Any]] = [] + + def run_crew() -> CrewOutput | None: + self.stream = False + try: + crew_result = self.kickoff(inputs=inputs, input_files=input_files) + return crew_result if isinstance(crew_result, CrewOutput) else None + finally: + self.stream = True + + stream_session: StreamSession[CrewOutput] = StreamSession( + sync_iterator=create_frame_generator(state, run_crew, output_holder) + ) + output_holder.append(stream_session) + return stream_session + + async def _akickoff_frame_stream( + self, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + ) -> AsyncStreamSession[CrewOutput]: + """Run kickoff asynchronously and stream full execution frames. + + Async counterpart of ``_kickoff_frame_stream``. Yields ``StreamFrame`` + events via the existing async frame streaming pipeline, instead of the + default ``StreamChunk`` token-only stream. Requires ``stream=True``; + ``stream_frames`` selects this path over chunk streaming. + + Args: + inputs: Optional input dictionary for task interpolation. + input_files: Optional dict of named file inputs for the crew. + + Returns: + An ``AsyncStreamSession`` yielding ``StreamFrame`` events. + """ + result_holder: list[CrewOutput] = [] + state = create_frame_streaming_state(result_holder, use_async=True) + output_holder: list[AsyncStreamSession[Any]] = [] + + async def run_crew() -> CrewOutput | None: + self.stream = False + try: + result = await asyncio.to_thread(self.kickoff, inputs, input_files) + return result if isinstance(result, CrewOutput) else None + finally: + self.stream = True + + stream_session: AsyncStreamSession[CrewOutput] = AsyncStreamSession( + async_iterator=create_async_frame_generator(state, run_crew, output_holder) + ) + output_holder.append(stream_session) + return stream_session + async def kickoff_for_each_async( self, inputs: list[dict[str, Any]], From 1689a923f5342c48a3cdf9783aa3cea301ea3d68 Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Fri, 3 Jul 2026 20:49:36 +0800 Subject: [PATCH 2/3] test(streaming): cover Crew stream_frames frame streaming Add TestCrewFrameStreaming covering: mixed-event streaming (asserts non-LLM 'tools' channel frames pass through, which the chunk path drops), result accessibility after exhaustion, default stream_frames=False compatibility, and the async _akickoff_frame_stream path. --- lib/crewai/tests/test_streaming.py | 153 +++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) diff --git a/lib/crewai/tests/test_streaming.py b/lib/crewai/tests/test_streaming.py index e3d84f4c99..635fbd911b 100644 --- a/lib/crewai/tests/test_streaming.py +++ b/lib/crewai/tests/test_streaming.py @@ -10,6 +10,7 @@ from crewai import Agent, Crew, Task from crewai.events.event_bus import crewai_event_bus from crewai.events.types.llm_events import LLMStreamChunkEvent, ToolCall, FunctionCall +from crewai.events.types.tool_usage_events import ToolUsageStartedEvent from crewai.flow.flow import Flow, start from crewai.state.checkpoint_config import CheckpointConfig from crewai.types.streaming import ( @@ -1011,3 +1012,155 @@ def emit_chunks(prefix: str, call_id: str) -> None: _unregister_handler(state_a.handler) _unregister_handler(state_b.handler) + + +# --------------------------------------------------------------------------- +# Frame-streaming (stream_frames) tests +# --------------------------------------------------------------------------- + + +class TestCrewFrameStreaming: + """Tests for Crew-level frame streaming (stream_frames=True).""" + + @staticmethod + def _fake_kickoff_returning( + output: Any, + events: list | None = None, + ) -> Any: + """Factory: returns a fake ``kickoff(self, ...)`` that emits *events* + on the crew bus and returns *output*.""" + + def fake_kickoff( + _self: Crew, + inputs: dict[str, Any] | None = None, + input_files: dict[str, Any] | None = None, + from_checkpoint: CheckpointConfig | None = None, + ) -> Any: + if events: + for event in events: + crewai_event_bus.emit(_self, event) + return output + + return fake_kickoff + + def test_frame_stream_yields_mixed_events( + self, researcher: Agent, simple_task: Task + ) -> None: + """When stream_frames=True, the StreamSession includes non-LLM events.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + fake_output.raw = "mock output" + + fake_kickoff = self._fake_kickoff_returning( + fake_output, + events=[ + LLMStreamChunkEvent( + type="llm_stream_chunk", chunk="token", call_id="c-1" + ), + ToolUsageStartedEvent( + type="tool_usage_started", + tool_name="search", + tool_args={"q": "test"}, + ), + ], + ) + + with patch.object(Crew, "kickoff", fake_kickoff): + session = crew._kickoff_frame_stream(None, None) + frames = list(session) + + assert len(frames) >= 2, f"expected >=2 frames, got {len(frames)}" + + channels = {f.channel for f in frames} + assert "llm" in channels, "LLM chunks should produce 'llm' channel frames" + assert ( + "tools" in channels + ), "tool events should produce 'tools' channel frames — " + "this proves the frame path includes non-LLM events that the chunk " + "path drops" + + assert session.result is fake_output + + def test_frame_stream_result_accessible_after_exhaustion( + self, researcher: Agent, simple_task: Task + ) -> None: + """StreamSession.result returns the CrewOutput after all frames are consumed.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + + with patch.object(Crew, "kickoff", self._fake_kickoff_returning(fake_output)): + session = crew._kickoff_frame_stream(None, None) + list(session) + assert session.result is fake_output + assert session.is_completed + + def test_stream_frames_false_default_unchanged( + self, researcher: Agent, simple_task: Task + ) -> None: + """stream_frames=False (default) — the Crew field is False out of the box.""" + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + ) + assert not crew.stream_frames + # The chunk path is still available (verified via existing tests). + + @pytest.mark.asyncio + async def test_async_frame_stream_yields_tool_events( + self, researcher: Agent, simple_task: Task + ) -> None: + """Async frame streaming via _akickoff_frame_stream yields tool frames.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + fake_output.raw = "mock async" + + fake_kickoff = self._fake_kickoff_returning( + fake_output, + events=[ + LLMStreamChunkEvent( + type="llm_stream_chunk", + chunk="async-token", + call_id="c-async", + ), + ToolUsageStartedEvent( + type="tool_usage_started", + tool_name="fetch", + tool_args={"url": "/api"}, + ), + ], + ) + + with patch.object(Crew, "kickoff", fake_kickoff): + session = await crew._akickoff_frame_stream(None, None) + frames = [f async for f in session] + + assert len(frames) >= 2 + channels = {f.channel for f in frames} + assert "tools" in channels + assert session.result is fake_output From 3910252bd25a45a2c31578a27b6913231cc43aae Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Sat, 4 Jul 2026 16:10:23 +0800 Subject: [PATCH 3/3] fix(crew): enforce stream_frames implies stream + fix test assert message Address CodeRabbit review: (1) add a model_validator so stream_frames=True auto-enables stream=True instead of being silently ignored when stream=False, and update the field docstring; (2) fix the split assertion in test_frame_stream_yields_mixed_events so the full failure message is part of the assert (the trailing strings were no-op expressions). Add a test covering the stream_frames->stream implication. --- lib/crewai/src/crewai/crew.py | 23 ++++++++++++++++++----- lib/crewai/tests/test_streaming.py | 23 ++++++++++++++++++----- 2 files changed, 36 insertions(+), 10 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index b5db421f9e..3ce54b1b81 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -303,11 +303,12 @@ class Crew(FlowTrackable, BaseModel): stream_frames: bool = Field( default=False, description=( - "When True (requires stream=True), kickoff yields StreamFrame " - "events covering the full execution (LLM tokens, tool calls, " - "agent/task lifecycle) instead of StreamChunk token chunks. " - "Returns a StreamSession / AsyncStreamSession whose .subscribe() " - "allows filtering by channel (llm, tools, messages, flow, lifecycle)." + "When True, kickoff yields StreamFrame events covering the full " + "execution (LLM tokens, tool calls, agent/task lifecycle) instead " + "of StreamChunk token chunks. Implies stream=True (streaming is " + "auto-enabled if not set). Returns a StreamSession / " + "AsyncStreamSession whose .subscribe() allows filtering by channel " + "(llm, tools, messages, flow, lifecycle)." ), ) max_rpm: int | None = Field( @@ -709,6 +710,18 @@ def create_crew_knowledge(self) -> Crew: ) return self + @model_validator(mode="after") + def enforce_stream_frames_contract(self) -> Self: + """Ensure ``stream_frames`` implies ``stream``. + + ``stream_frames`` only takes effect when ``stream`` is True. Rather + than silently ignoring ``stream_frames=True`` with ``stream=False``, + enable streaming so the frame output is produced as documented. + """ + if self.stream_frames and not self.stream: + self.stream = True + return self + @model_validator(mode="after") def check_manager_llm(self) -> Self: """Validates that the language model is set when using hierarchical process.""" diff --git a/lib/crewai/tests/test_streaming.py b/lib/crewai/tests/test_streaming.py index 635fbd911b..3c3ad5a65a 100644 --- a/lib/crewai/tests/test_streaming.py +++ b/lib/crewai/tests/test_streaming.py @@ -1081,11 +1081,11 @@ def test_frame_stream_yields_mixed_events( channels = {f.channel for f in frames} assert "llm" in channels, "LLM chunks should produce 'llm' channel frames" - assert ( - "tools" in channels - ), "tool events should produce 'tools' channel frames — " - "this proves the frame path includes non-LLM events that the chunk " - "path drops" + assert "tools" in channels, ( + "tool events should produce 'tools' channel frames — " + "this proves the frame path includes non-LLM events that the chunk " + "path drops" + ) assert session.result is fake_output @@ -1110,6 +1110,19 @@ def test_frame_stream_result_accessible_after_exhaustion( assert session.result is fake_output assert session.is_completed + def test_stream_frames_implies_stream( + self, researcher: Agent, simple_task: Task + ) -> None: + """stream_frames=True auto-enables stream=True (no silent ignore).""" + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream_frames=True, + ) + assert crew.stream is True + assert crew.stream_frames is True + def test_stream_frames_false_default_unchanged( self, researcher: Agent, simple_task: Task ) -> None: