feat(crew): add stream_frames opt-in for full event frame streaming

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,17 @@ class Crew(FlowTrackable, BaseModel):
         default=False,
         description="Whether to stream output from the crew execution.",
     )
+    stream_frames: bool = Field(
+        default=False,
+        description=(
+            "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(
         default=None,
         description=(
@@ -692,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."""
@@ -968,7 +998,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 +1017,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 +1130,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 +1153,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 +1181,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]],

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,168 @@ 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_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:
+        """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