Merge branch 'main' into fix/output-file-interpolation-traversal

nltk 3.9.4 path-traversal advisory (percent-encoded sequences bypassing
the data.load()/find() check; incomplete fix for nltk#3504). 3.9.4 is the
latest release and the advisory lists no fixed version, so the pin cannot be
bumped to clear it. nltk is transitive via unstructured and crewai never
calls nltk.data.load()/find() with untrusted input. Mirrors the existing
PYSEC-2026-97 nltk ignore.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/workflows/vulnerability-scan.yml

@@ -51,6 +51,7 @@ jobs:
             --ignore-vuln PYSEC-2024-277 \
             --ignore-vuln PYSEC-2026-89 \
             --ignore-vuln PYSEC-2026-97 \
+            --ignore-vuln PYSEC-2026-597 \
             --ignore-vuln PYSEC-2025-148 \
             --ignore-vuln PYSEC-2025-183 \
             --ignore-vuln PYSEC-2025-189 \
@@ -78,6 +79,7 @@ jobs:
         #   PYSEC-2024-277      - joblib 1.5.3: disputed; NumpyArrayWrapper only used with trusted caches
         #   PYSEC-2026-89       - markdown 3.10.2: DoS via malformed HTML; fix 3.8.1 — already past, advisory range is stale
         #   PYSEC-2026-97       - nltk 3.9.4: arbitrary file read in filestring(); no fix available
+        #   PYSEC-2026-597      - nltk 3.9.4: path traversal via percent-encoded sequences in data.load()/find() (incomplete fix for #3504); latest release, no fix available. Transitive via unstructured; crewai never calls nltk.data.load/find with untrusted input.
         #   PYSEC-2025-148      - onnx 1.21.0: path traversal in save_external_data; no fix available
         #   PYSEC-2025-183      - pyjwt 2.12.1: disputed weak-encryption claim; key length is application-chosen
         #   PYSEC-2025-189..197 - torch 2.11.0: memory-corruption/DoS in functions only reachable via untrusted models; no fix available

docs/edge/en/concepts/streaming.mdx

@@ -61,35 +61,6 @@ Frames are grouped into high-level channels:
 
 The stream itself remains one ordered timeline. Channel projections let consumers focus on only part of that timeline.
 
-### Tool Call Streaming
-
-Tool calls appear in two places because there are two distinct phases:
-
-| Phase | Channel | Event type | Meaning |
-|-------|---------|------------|---------|
-| Tool-call construction | `llm` | `llm_stream_chunk` | The model is streaming the tool name or arguments it intends to call |
-| Tool execution | `tools` | `tool_usage_started`, `tool_usage_finished`, `tool_usage_error` | CrewAI is executing the tool and reporting the result |
-
-For streamed tool-call arguments, the latest provider delta is available as `frame.data["chunk"]`. The accumulated argument string is available at `frame.data["tool_call"]["function"]["arguments"]`.
-
-```python
-with llm.stream_events("Check the weather in Paris.", tools=[weather_tool]) as stream:
-    for frame in stream.llm:
-        if frame.type != "llm_stream_chunk":
-            continue
-
-        tool_call = frame.data.get("tool_call")
-        if tool_call:
-            function = tool_call["function"]
-            print("Tool:", function["name"])
-            print("Latest argument delta:", frame.data["chunk"])
-            print("Arguments so far:", function["arguments"])
-        elif frame.content:
-            print(frame.content, end="", flush=True)
-```
-
-Use the `tools` channel when you want to display that a tool actually started, completed, or failed. Use the `llm` channel when you want to observe the model constructing the tool call before execution.
-
 ```mermaid
 flowchart LR
     A["flow<br/>flow_started"] --> B["llm<br/>llm_call_started"]

docs/edge/en/learn/consuming-streams.mdx

@@ -70,38 +70,6 @@ result = stream.result
 
 `frame.event` is the structured payload for the source event. Use it for metadata such as tool names, arguments, message roles, and runtime identifiers.
 
-## Print Streamed Tool Call Arguments
-
-When a model supports streamed tool calling, CrewAI emits partial tool-call arguments as `llm_stream_chunk` frames on the `llm` channel. These frames are different from tool execution events:
-
-- `llm` channel: the model is constructing the tool call.
-- `tools` channel: CrewAI is executing the tool.
-
-The current delta is stored in `frame.event["chunk"]`. The accumulated arguments are stored in `frame.event["tool_call"]["function"]["arguments"]`.
-
-```python
-with llm.stream_events("Get the weather in Paris.", tools=[weather_tool]) as stream:
-    for frame in stream.interleave(["llm", "tools"]):
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            tool_call = frame.event.get("tool_call")
-            if tool_call:
-                function = tool_call["function"]
-                print(f"\nPreparing tool: {function['name']}")
-                print(f"Arguments so far: {function['arguments']}")
-            elif frame.content:
-                print(frame.content, end="", flush=True)
-
-        elif frame.channel == "tools" and frame.type == "tool_usage_started":
-            print(f"\nRunning tool: {frame.event.get('tool_name')}")
-
-        elif frame.channel == "tools" and frame.type == "tool_usage_finished":
-            print(f"\nTool finished: {frame.event.get('tool_name')}")
-
-result = stream.result
-```
-
-Some providers emit arguments in small JSON fragments. For display, prefer the accumulated argument string over the latest delta.
-
 ## Watch Flow Progress
 
 Flow lifecycle and method execution frames arrive on the `flow` channel:

docs/edge/en/learn/streaming-crew-execution.mdx

@@ -240,18 +240,15 @@ for chunk in streaming:
 
 ### TOOL_CALL Chunks
 
-Information about tool calls being made. Depending on the provider, CrewAI may receive tool-call arguments incrementally. In that case, each `TOOL_CALL` chunk contains the latest streamed content in `chunk.content`, while `chunk.tool_call.arguments` contains the accumulated argument string so far.
+Information about tool calls being made:
 
 ```python Code
 for chunk in streaming:
     if chunk.chunk_type == StreamChunkType.TOOL_CALL:
         print(f"\nCalling tool: {chunk.tool_call.tool_name}")
-        print(f"Latest argument delta: {chunk.content}")
-        print(f"Arguments so far: {chunk.tool_call.arguments}")
+        print(f"Arguments: {chunk.tool_call.arguments}")
 ```
 
-Actual tool execution is reported separately through tool usage events in the runtime event stream and through CrewAI's verbose console output. `TOOL_CALL` chunks represent the model constructing the tool request before or during execution.
-
 ## Practical Example: Building a UI with Streaming
 
 Here's a complete example showing how to build an interactive application with streaming:
@@ -384,4 +381,4 @@ except Exception as e:
         print("Streaming completed but an error occurred")
 ```
 
-By leveraging streaming, you can build more responsive and interactive applications with CrewAI, providing users with real-time visibility into agent execution and results.
+By leveraging streaming, you can build more responsive and interactive applications with CrewAI, providing users with real-time visibility into agent execution and results.

docs/edge/en/learn/streaming-runtime-contract.mdx

@@ -145,33 +145,6 @@ result = stream.result
 
 `llm.stream_events(...)` temporarily enables streaming for the wrapped call and restores the LLM's previous `stream` setting afterward. Provider integrations continue to emit the underlying LLM stream events; this helper provides a common iterator API over those events for every LLM provider.
 
-### Tool Call Argument Deltas
-
-Native tool-call streaming uses `llm_stream_chunk` frames on the `llm` channel. This represents the model constructing a tool call. It is separate from the `tools` channel, which represents CrewAI executing that tool.
-
-For a streamed tool call, the frame payload includes:
-
-```python
-frame.type == "llm_stream_chunk"
-frame.channel == "llm"
-frame.event["call_type"] == "tool_call"
-frame.event["chunk"]      # latest argument delta from the provider
-frame.event["tool_call"]  # accumulated tool-call state
-```
-
-The `tool_call` payload follows the OpenAI-style function call shape:
-
-```python
-tool_call = frame.event["tool_call"]
-tool_call["id"]
-tool_call["index"]
-tool_call["type"]  # "function"
-tool_call["function"]["name"]
-tool_call["function"]["arguments"]  # accumulated JSON argument string
-```
-
-Providers differ in granularity. OpenAI and Anthropic may stream arguments as small JSON fragments, while some providers emit the complete argument object in one chunk. Consumers should treat `frame.event["chunk"]` as the latest delta and `tool_call["function"]["arguments"]` as the current accumulated state.
-
 ## Conversational Turns
 
 Conversational Flows can stream one user turn with `stream_turn()`:

lib/cli/tests/test_plus_api.py

@@ -1,8 +1,6 @@
 import os
 import unittest
-from unittest.mock import ANY, AsyncMock, MagicMock, patch
-
-import pytest
+from unittest.mock import ANY, MagicMock, patch
 
 from crewai_cli.plus_api import PlusAPI
 
@@ -343,28 +341,23 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
-async def test_get_agent(mock_async_client_class):
+@patch("crewai_core.plus_api.PlusAPI._make_request")
+def test_get_agent(mock_make_request):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert response == mock_response
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
+@patch("crewai_core.plus_api.PlusAPI._make_request")
 @patch("crewai_core.plus_api.Settings")
-async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
+def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -374,15 +367,12 @@ async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_cl
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai-core/src/crewai_core/plus_api.py

@@ -232,10 +232,8 @@ class PlusAPI:
     def get_tool(self, handle: str) -> httpx.Response:
         return self._make_request("GET", f"{self.TOOLS_RESOURCE}/{handle}")
 
-    async def get_agent(self, handle: str) -> httpx.Response:
-        url = urljoin(self.base_url, f"{self.AGENTS_RESOURCE}/{handle}")
-        async with httpx.AsyncClient() as client:
-            return await client.get(url, headers=cast(dict[str, str], self.headers))
+    def get_agent(self, handle: str) -> httpx.Response:
+        return self._make_request("GET", f"{self.AGENTS_RESOURCE}/{handle}")
 
     def publish_tool(
         self,

lib/crewai/src/crewai/events/event_listener.py

@@ -64,7 +64,6 @@ from crewai.events.types.llm_events import (
     LLMCallCompletedEvent,
     LLMCallFailedEvent,
     LLMCallStartedEvent,
-    LLMCallType,
     LLMStreamChunkEvent,
 )
 from crewai.events.types.llm_guardrail_events import (
@@ -456,13 +455,6 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(LLMStreamChunkEvent)
         def on_llm_stream_chunk(_: Any, event: LLMStreamChunkEvent) -> None:
-            if event.call_type == LLMCallType.TOOL_CALL:
-                self.formatter.handle_llm_stream_chunk(
-                    event.tool_call.function.arguments if event.tool_call else "",
-                    event.call_type,
-                )
-                return
-
             self.text_stream.write(event.chunk)
             self.text_stream.seek(self.next_chunk)
             self.text_stream.read()

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -437,8 +437,6 @@ To enable tracing, do any one of these:
         if not self.verbose:
             return
 
-        self.pause_live_updates()
-
         with self._tool_counts_lock:
             self.tool_usage_counts[tool_name] = (
                 self.tool_usage_counts.get(tool_name, 0) + 1
@@ -470,8 +468,6 @@ To enable tracing, do any one of these:
         if not self.verbose:
             return
 
-        self.pause_live_updates()
-
         with self._tool_counts_lock:
             iteration = self.tool_usage_counts.get(tool_name, 1)
 
@@ -499,8 +495,6 @@ To enable tracing, do any one of these:
         if not self.verbose:
             return
 
-        self.pause_live_updates()
-
         with self._tool_counts_lock:
             iteration = self.tool_usage_counts.get(tool_name, 1)
 
@@ -549,15 +543,6 @@ To enable tracing, do any one of these:
         if should_suppress_console_output():
             return
 
-        from crewai.events.types.llm_events import LLMCallType
-
-        if call_type == LLMCallType.TOOL_CALL:
-            self.pause_live_updates()
-            self._is_streaming = False
-            self._last_stream_call_type = call_type
-            self._just_streamed_final_answer = False
-            return
-
         self._is_streaming = True
         self._last_stream_call_type = call_type
 
@@ -569,9 +554,17 @@ To enable tracing, do any one of these:
             display_text = "...\n" + display_text
 
         content = Text()
-        content.append(display_text, style="bright_green")
-        title = "✅ Agent Final Answer"
-        border_style = "green"
+
+        from crewai.events.types.llm_events import LLMCallType
+
+        if call_type == LLMCallType.TOOL_CALL:
+            content.append(display_text, style="yellow")
+            title = "🔧 Tool Arguments"
+            border_style = "yellow"
+        else:
+            content.append(display_text, style="bright_green")
+            title = "✅ Agent Final Answer"
+            border_style = "green"
 
         streaming_panel = Panel(
             content,

lib/crewai/src/crewai/flow/runtime/_actions.py

@@ -138,11 +138,12 @@ class CrewAction:
 
         local_context = _pop_local_context(kwargs)
         if self.definition.from_declaration is not None:
-            crew, default_inputs = load_crew(
+            crew, default_inputs = await asyncio.to_thread(
+                load_crew,
                 _resolve_crew_declaration(
                     self.definition.from_declaration,
                     base_dir=self.flow._definition.source_dir,
-                )
+                ),
             )
             input_template = {**default_inputs, **(self.definition.inputs or {})}
         else:
@@ -155,7 +156,9 @@ class CrewAction:
                 **crew_definition.inputs,
                 **(self.definition.inputs or {}),
             }
-            crew, _ = load_crew_from_definition(crew_definition, source="crew action")
+            crew, _ = await asyncio.to_thread(
+                load_crew_from_definition, crew_definition, source="crew action"
+            )
 
         inputs = Expression.from_flow(
             cast(ExpressionData, input_template),
@@ -184,7 +187,8 @@ class AgentAction:
         if not isinstance(rendered_input, str):
             raise ValueError("agent input must render to a string")
 
-        agent, response_format = load_agent_from_definition(
+        agent, response_format = await asyncio.to_thread(
+            load_agent_from_definition,
             self.definition.with_,
             source="agent action",
         )

lib/crewai/src/crewai/llms/base_llm.py

@@ -685,40 +685,6 @@ class BaseLLM(BaseModel, ABC):
             ),
         )
 
-    def _emit_tool_call_stream_chunk_event(
-        self,
-        *,
-        chunk: str,
-        tool_call_id: str | None,
-        tool_name: str | None,
-        arguments: str,
-        index: int,
-        from_task: Task | None = None,
-        from_agent: BaseAgent | None = None,
-        response_id: str | None = None,
-    ) -> None:
-        """Emit a normalized streamed tool-call chunk.
-
-        ``chunk`` is the latest provider delta while ``arguments`` is the
-        accumulated argument string for consumers that want current state.
-        """
-        self._emit_stream_chunk_event(
-            chunk=chunk,
-            from_task=from_task,
-            from_agent=from_agent,
-            tool_call={
-                "id": tool_call_id,
-                "function": {
-                    "name": tool_name or "",
-                    "arguments": arguments,
-                },
-                "type": "function",
-                "index": index,
-            },
-            call_type=LLMCallType.TOOL_CALL,
-            response_id=response_id,
-        )
-
     def _emit_thinking_chunk_event(
         self,
         chunk: str,

lib/crewai/src/crewai/llms/providers/openai/completion.py

@@ -1102,7 +1102,6 @@ class OpenAICompletion(BaseLLM):
         """Handle streaming Responses API call."""
         full_response = ""
         function_calls: list[dict[str, Any]] = []
-        streaming_function_calls: dict[str, dict[str, Any]] = {}
         final_response: Response | None = None
         usage: dict[str, Any] | None = None
 
@@ -1124,46 +1123,11 @@ class OpenAICompletion(BaseLLM):
                 )
 
             elif event.type == "response.function_call_arguments.delta":
-                self._process_responses_function_call_delta(
-                    event=event,
-                    function_calls=streaming_function_calls,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_id=response_id_stream,
-                )
-
-            elif event.type == "response.function_call_arguments.done":
-                self._process_responses_function_call_done(
-                    event=event,
-                    function_calls=streaming_function_calls,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_id=response_id_stream,
-                )
-
-            elif event.type == "response.output_item.added":
-                item = event.item
-                if item.type == "function_call":
-                    self._process_responses_function_call_started(
-                        event=event,
-                        function_calls=streaming_function_calls,
-                        from_task=from_task,
-                        from_agent=from_agent,
-                        response_id=response_id_stream,
-                    )
+                pass
 
             elif event.type == "response.output_item.done":
                 item = event.item
                 if item.type == "function_call":
-                    call_key = self._responses_function_call_key(event)
-                    if call_key in streaming_function_calls:
-                        streaming_function_calls[call_key].update(
-                            {
-                                "id": item.call_id,
-                                "name": item.name,
-                                "arguments": item.arguments,
-                            }
-                        )
                     function_calls.append(
                         {
                             "id": item.call_id,
@@ -1275,7 +1239,6 @@ class OpenAICompletion(BaseLLM):
         """Handle async streaming Responses API call."""
         full_response = ""
         function_calls: list[dict[str, Any]] = []
-        streaming_function_calls: dict[str, dict[str, Any]] = {}
         final_response: Response | None = None
         usage: dict[str, Any] | None = None
 
@@ -1297,46 +1260,11 @@ class OpenAICompletion(BaseLLM):
                 )
 
             elif event.type == "response.function_call_arguments.delta":
-                self._process_responses_function_call_delta(
-                    event=event,
-                    function_calls=streaming_function_calls,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_id=response_id_stream,
-                )
-
-            elif event.type == "response.function_call_arguments.done":
-                self._process_responses_function_call_done(
-                    event=event,
-                    function_calls=streaming_function_calls,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_id=response_id_stream,
-                )
-
-            elif event.type == "response.output_item.added":
-                item = event.item
-                if item.type == "function_call":
-                    self._process_responses_function_call_started(
-                        event=event,
-                        function_calls=streaming_function_calls,
-                        from_task=from_task,
-                        from_agent=from_agent,
-                        response_id=response_id_stream,
-                    )
+                pass
 
             elif event.type == "response.output_item.done":
                 item = event.item
                 if item.type == "function_call":
-                    call_key = self._responses_function_call_key(event)
-                    if call_key in streaming_function_calls:
-                        streaming_function_calls[call_key].update(
-                            {
-                                "id": item.call_id,
-                                "name": item.name,
-                                "arguments": item.arguments,
-                            }
-                        )
                     function_calls.append(
                         {
                             "id": item.call_id,
@@ -1435,135 +1363,6 @@ class OpenAICompletion(BaseLLM):
 
         return full_response
 
-    @staticmethod
-    def _responses_function_call_key(event: Any) -> str:
-        item = getattr(event, "item", None)
-        item_id = getattr(event, "item_id", None) or getattr(item, "id", None)
-        if item_id:
-            return str(item_id)
-        output_index = getattr(event, "output_index", None)
-        if output_index is not None:
-            return f"output:{output_index}"
-        return "output:0"
-
-    def _process_responses_function_call_delta(
-        self,
-        *,
-        event: Any,
-        function_calls: dict[str, dict[str, Any]],
-        from_task: Any | None = None,
-        from_agent: Any | None = None,
-        response_id: str | None = None,
-    ) -> None:
-        call_key = self._responses_function_call_key(event)
-        output_index = getattr(event, "output_index", 0) or 0
-        item_id = getattr(event, "item_id", None)
-        delta = getattr(event, "delta", "") or ""
-        call_data = function_calls.setdefault(
-            call_key,
-            {
-                "id": item_id,
-                "name": "",
-                "arguments": "",
-                "index": output_index,
-                "streamed_arguments": False,
-            },
-        )
-        if item_id and not call_data.get("id"):
-            call_data["id"] = item_id
-        call_data["arguments"] += delta
-        call_data["streamed_arguments"] = True
-
-        self._emit_tool_call_stream_chunk_event(
-            chunk=delta,
-            tool_call_id=call_data.get("id"),
-            tool_name=call_data.get("name"),
-            arguments=call_data["arguments"],
-            index=call_data["index"],
-            from_task=from_task,
-            from_agent=from_agent,
-            response_id=response_id,
-        )
-
-    def _process_responses_function_call_started(
-        self,
-        *,
-        event: Any,
-        function_calls: dict[str, dict[str, Any]],
-        from_task: Any | None = None,
-        from_agent: Any | None = None,
-        response_id: str | None = None,
-    ) -> None:
-        call_key = self._responses_function_call_key(event)
-        output_index = getattr(event, "output_index", 0) or 0
-        item = getattr(event, "item", None)
-        item_id = getattr(item, "id", None)
-        call_data = function_calls.setdefault(
-            call_key,
-            {
-                "id": getattr(item, "call_id", None) or item_id,
-                "name": getattr(item, "name", None) or "",
-                "arguments": "",
-                "index": output_index,
-                "streamed_arguments": False,
-            },
-        )
-        call_data["id"] = call_data.get("id") or getattr(item, "call_id", None)
-        call_data["name"] = call_data.get("name") or getattr(item, "name", None) or ""
-
-        self._emit_tool_call_stream_chunk_event(
-            chunk="",
-            tool_call_id=call_data.get("id"),
-            tool_name=call_data.get("name"),
-            arguments=call_data["arguments"],
-            index=call_data["index"],
-            from_task=from_task,
-            from_agent=from_agent,
-            response_id=response_id,
-        )
-
-    def _process_responses_function_call_done(
-        self,
-        *,
-        event: Any,
-        function_calls: dict[str, dict[str, Any]],
-        from_task: Any | None = None,
-        from_agent: Any | None = None,
-        response_id: str | None = None,
-    ) -> None:
-        call_key = self._responses_function_call_key(event)
-        output_index = getattr(event, "output_index", 0) or 0
-        item_id = getattr(event, "item_id", None)
-        arguments = getattr(event, "arguments", "") or ""
-        call_data = function_calls.setdefault(
-            call_key,
-            {
-                "id": item_id,
-                "name": "",
-                "arguments": "",
-                "index": output_index,
-                "streamed_arguments": False,
-            },
-        )
-        if item_id and not call_data.get("id"):
-            call_data["id"] = item_id
-        call_data["name"] = getattr(event, "name", None) or call_data.get("name", "")
-
-        if not call_data.get("streamed_arguments") and arguments:
-            call_data["arguments"] = arguments
-            self._emit_tool_call_stream_chunk_event(
-                chunk=arguments,
-                tool_call_id=call_data.get("id"),
-                tool_name=call_data.get("name"),
-                arguments=call_data["arguments"],
-                index=call_data["index"],
-                from_task=from_task,
-                from_agent=from_agent,
-                response_id=response_id,
-            )
-        elif arguments:
-            call_data["arguments"] = arguments
-
     def _extract_function_calls_from_response(
         self, response: Response
     ) -> list[dict[str, Any]]:

lib/crewai/src/crewai/project/crew_definition.py

@@ -66,21 +66,24 @@ class CrewAgentDefinition(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    role: str = Field(
+    role: str | None = Field(
+        default=None,
         description=(
             "Crew agent role. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
         ),
         examples=["Research analyst"],
     )
-    goal: str = Field(
+    goal: str | None = Field(
+        default=None,
         description=(
             "Crew agent goal. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
         ),
         examples=["Research {topic}"],
     )
-    backstory: str = Field(
+    backstory: str | None = Field(
+        default=None,
         description=(
             "Crew agent backstory. Crew inputs are interpolated with `{name}` "
             "placeholders such as `{topic}`; this is not CEL."
@@ -92,6 +95,15 @@ class CrewAgentDefinition(BaseModel):
         description="Optional built-in type or Python reference used to load the agent.",
         examples=["agent", {"python": "my_project.agents.ResearchAgent"}],
     )
+    from_repository: str | None = Field(
+        default=None,
+        description=(
+            "Agent repository name to load. Repository values supply missing "
+            "agent configuration; explicitly provided local fields override the "
+            "repository values."
+        ),
+        examples=["researcher"],
+    )
     settings: dict[str, Any] = Field(
         default_factory=dict,
         description="Additional agent settings passed to the loader.",
@@ -183,15 +195,18 @@ class CrewAgentDefinition(BaseModel):
 class AgentDefinition(CrewAgentDefinition):
     """Inline individual agent definition used outside of a crew."""
 
-    role: str = Field(
+    role: str | None = Field(
+        default=None,
         description="Individual agent role used by a Flow agent action outside of a crew.",
         examples=["Support specialist"],
     )
-    goal: str = Field(
+    goal: str | None = Field(
+        default=None,
         description="Individual agent goal for the Flow agent action outside of a crew.",
         examples=["Draft a concise customer reply"],
     )
-    backstory: str = Field(
+    backstory: str | None = Field(
+        default=None,
         description=(
             "Individual agent backstory used to shape behavior outside of a crew."
         ),

lib/crewai/src/crewai/project/json_loader.py

@@ -978,9 +978,10 @@ def _agent_kwargs_from_definition(
         extra_allowed,
         skip_unknown=skip_unknown,
     )
-    for required in ("role", "goal", "backstory"):
-        if required not in defn:
-            errors.append(f"{path}: missing required field '{required}'")
+    if not defn.get("from_repository"):
+        for required in ("role", "goal", "backstory"):
+            if defn.get(required) is None:
+                errors.append(f"{path}: missing required field '{required}'")
 
     settings = defn.get("settings", {})
     if settings is None:

lib/crewai/src/crewai/task.py

@@ -10,7 +10,7 @@ from hashlib import md5
 import inspect
 import json
 import logging
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 import threading
 from typing import (
     Annotated,
@@ -504,6 +504,29 @@ class Task(BaseModel):
         if value is None:
             return None
 
+        if "{" in value or "}" in value:
+            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
+            for var in template_vars:
+                if not var.isidentifier():
+                    raise ValueError(f"Invalid template variable name: {var}")
+            # Literal portions are still checked here; the fully interpolated
+            # path is re-validated at runtime (see
+            # interpolate_inputs_and_add_conversation_history) because template
+            # variables may be filled from untrusted kickoff inputs.
+            cls._sanitize_output_file_path(value)
+            return value
+
+        return cls._sanitize_output_file_path(value)
+
+    @staticmethod
+    def _sanitize_output_file_path(value: str) -> str:
+        """Enforce path-safety on an ``output_file`` value.
+
+        Shared by the field validator's literal and template branches. Rejects
+        traversal sequences, shell expansion, and shell metacharacters, and
+        strips a leading ``/`` so a literal path stays relative to the working
+        directory.
+        """
         if ".." in value:
             raise ValueError(
                 "Path traversal attempts are not allowed in output_file paths"
@@ -519,17 +542,56 @@ class Task(BaseModel):
                 "Shell special characters are not allowed in output_file paths"
             )
 
-        if "{" in value or "}" in value:
-            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
-            for var in template_vars:
-                if not var.isidentifier():
-                    raise ValueError(f"Invalid template variable name: {var}")
-            return value
-
         if value.startswith("/"):
             return value[1:]
         return value
 
+    def _validate_output_file_input_values(
+        self, inputs: dict[str, str | int | float | dict[str, Any] | list[Any]]
+    ) -> None:
+        """Reject untrusted input values that would escape the output path.
+
+        Only the variables that actually appear in the ``output_file`` template
+        are checked. The developer-authored template is trusted (it may contain
+        an absolute base directory), but a value substituted into it must not
+        introduce path traversal (``..``), an absolute path, a home/variable
+        expansion (``~``/``$``), or shell metacharacters that would redirect the
+        write outside the intended location.
+        """
+        if not self._original_output_file:
+            return
+
+        template_vars = [
+            part.split("}")[0] for part in self._original_output_file.split("{")[1:]
+        ]
+        for var in template_vars:
+            if var not in inputs:
+                continue
+            value = str(inputs[var])
+            if ".." in value:
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': path "
+                    "traversal sequences ('..') are not allowed"
+                )
+            if value.startswith(("~", "$")):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': shell "
+                    "expansion characters are not allowed"
+                )
+            if any(char in value for char in ["|", ">", "<", "&", ";"]):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': shell "
+                    "special characters are not allowed"
+                )
+            if (
+                PurePosixPath(value).is_absolute()
+                or PureWindowsPath(value).is_absolute()
+            ):
+                raise ValueError(
+                    f"Invalid value for output_file variable '{var}': absolute "
+                    "paths are not allowed"
+                )
+
     @model_validator(mode="after")
     def set_attributes_based_on_config(self) -> Task:
         """Set attributes based on the agent configuration."""
@@ -1026,6 +1088,12 @@ Follow these guidelines:
             raise ValueError(f"Error interpolating expected_output: {e!s}") from e
 
         if self.output_file is not None:
+            # Values interpolated into the output path may come from untrusted
+            # kickoff inputs. The developer-authored template (including any
+            # absolute base directory) is trusted, but an injected value must
+            # not introduce path traversal, an absolute path, or shell
+            # expansion that would escape the intended location.
+            self._validate_output_file_input_values(inputs)
             try:
                 self.output_file = interpolate_only(
                     input_string=self._original_output_file, inputs=inputs

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -1125,7 +1125,7 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
 
             client = PlusAPI(api_key=get_auth_token())
         _print_current_organization()
-        response = asyncio.run(client.get_agent(from_repository))
+        response = client.get_agent(from_repository)
         if response.status_code == 404:
             raise AgentRepositoryError(
                 f"Agent {from_repository} does not exist, make sure the name is correct or the agent is available on your organization."
@@ -1158,6 +1158,8 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
                         raise AgentRepositoryError(
                             f"Tool {tool['name']} could not be loaded: {e}"
                         ) from e
+            elif key == "skills" and value == []:
+                continue
             else:
                 attributes[key] = value
     return attributes

lib/crewai/tests/agents/test_agent.py

@@ -2243,6 +2243,27 @@ def test_agent_from_repository_override_attributes(mock_get_agent, mock_get_auth
     assert isinstance(agent.tools[0], SerperDevTool)
 
 
+@patch("crewai.plus_api.PlusAPI.get_agent")
+def test_agent_from_repository_ignores_empty_skills(
+    mock_get_agent, mock_get_auth_token
+):
+    mock_get_response = MagicMock()
+    mock_get_response.status_code = 200
+    mock_get_response.json.return_value = {
+        "role": "test role",
+        "goal": "test goal",
+        "backstory": "test backstory",
+        "tools": [],
+        "skills": [],
+    }
+    mock_get_agent.return_value = mock_get_response
+
+    agent = Agent(from_repository="test_agent")
+
+    assert agent.role == "test role"
+    assert agent.skills is None
+
+
 @patch("crewai.plus_api.PlusAPI.get_agent")
 def test_agent_from_repository_with_invalid_tools(mock_get_agent, mock_get_auth_token):
     mock_get_response = MagicMock()

lib/crewai/tests/cli/test_plus_api.py

@@ -1,8 +1,6 @@
 import os
 import unittest
-from unittest.mock import ANY, AsyncMock, MagicMock, patch
-
-import pytest
+from unittest.mock import ANY, MagicMock, patch
 
 from crewai.plus_api import PlusAPI
 
@@ -396,28 +394,23 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
-async def test_get_agent(mock_async_client_class):
+@patch("crewai_core.plus_api.PlusAPI._make_request")
+def test_get_agent(mock_make_request):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert response == mock_response
 
 
-@pytest.mark.asyncio
-@patch("httpx.AsyncClient")
+@patch("crewai_core.plus_api.PlusAPI._make_request")
 @patch("crewai_core.plus_api.Settings")
-async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
+def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -427,15 +420,12 @@ async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_cl
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_client_instance = AsyncMock()
-    mock_client_instance.get.return_value = mock_response
-    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
+    mock_make_request.return_value = mock_response
 
-    response = await api.get_agent("test_agent_handle")
+    response = api.get_agent("test_agent_handle")
 
-    mock_client_instance.get.assert_called_once_with(
-        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
-        headers=api.headers,
+    mock_make_request.assert_called_once_with(
+        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai/tests/llms/openai/test_openai.py

@@ -7,7 +7,6 @@ import openai
 import pytest
 
 from crewai.llm import LLM
-from crewai.events.types.llm_events import LLMCallType, LLMStreamChunkEvent
 from crewai.llms.providers.openai.completion import OpenAICompletion, ResponsesAPIResult
 from crewai.crew import Crew
 from crewai.agent import Agent
@@ -1888,198 +1887,6 @@ def test_openai_responses_api_no_detail_fields_omitted():
     assert "reasoning_tokens" not in usage
 
 
-def test_openai_responses_streaming_emits_tool_call_argument_deltas():
-    llm = OpenAICompletion(model="gpt-4o", api="responses", stream=True)
-    stream_events = [
-        types.SimpleNamespace(
-            type="response.created",
-            response=types.SimpleNamespace(id="resp_123"),
-        ),
-        types.SimpleNamespace(
-            type="response.output_item.added",
-            output_index=0,
-            item=types.SimpleNamespace(
-                type="function_call",
-                id="fc_123",
-                call_id="call_123",
-                name="get_weather",
-                arguments="",
-            ),
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.delta",
-            item_id="fc_123",
-            output_index=0,
-            delta='{"city"',
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.delta",
-            item_id="fc_123",
-            output_index=0,
-            delta=': "Paris"}',
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.done",
-            item_id="fc_123",
-            output_index=0,
-            name="get_weather",
-            arguments='{"city": "Paris"}',
-        ),
-        types.SimpleNamespace(
-            type="response.output_item.done",
-            output_index=0,
-            item=types.SimpleNamespace(
-                type="function_call",
-                id="fc_123",
-                call_id="call_123",
-                name="get_weather",
-                arguments='{"city": "Paris"}',
-            ),
-        ),
-        types.SimpleNamespace(
-            type="response.completed",
-            response=types.SimpleNamespace(
-                id="resp_123",
-                status="completed",
-                usage=None,
-            ),
-        ),
-    ]
-    fake_client = types.SimpleNamespace(
-        responses=types.SimpleNamespace(create=lambda **_: iter(stream_events))
-    )
-
-    with patch.object(llm, "_get_sync_client", return_value=fake_client):
-        with patch("crewai.events.event_bus.CrewAIEventsBus.emit") as mock_emit:
-            llm._handle_streaming_responses({"input": []})
-
-    tool_call_events = [
-        call.kwargs["event"]
-        for call in mock_emit.call_args_list
-        if isinstance(call.kwargs.get("event"), LLMStreamChunkEvent)
-        and call.kwargs["event"].call_type == LLMCallType.TOOL_CALL
-    ]
-
-    assert [event.chunk for event in tool_call_events] == [
-        "",
-        '{"city"',
-        ': "Paris"}',
-    ]
-    assert [
-        event.tool_call.function.arguments for event in tool_call_events
-    ] == ["", '{"city"', '{"city": "Paris"}']
-    assert all(event.tool_call.id == "call_123" for event in tool_call_events)
-    assert all(
-        event.tool_call.function.name == "get_weather" for event in tool_call_events
-    )
-
-
-@pytest.mark.asyncio
-async def test_openai_responses_async_streaming_emits_tool_call_argument_deltas():
-    llm = OpenAICompletion(model="gpt-4o", api="responses", stream=True)
-    stream_events = [
-        types.SimpleNamespace(
-            type="response.created",
-            response=types.SimpleNamespace(id="resp_123"),
-        ),
-        types.SimpleNamespace(
-            type="response.output_item.added",
-            output_index=0,
-            item=types.SimpleNamespace(
-                type="function_call",
-                id="fc_123",
-                call_id="call_123",
-                name="get_weather",
-                arguments="",
-            ),
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.delta",
-            item_id="fc_123",
-            output_index=0,
-            delta='{"city"',
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.delta",
-            item_id="fc_123",
-            output_index=0,
-            delta=': "Paris"}',
-        ),
-        types.SimpleNamespace(
-            type="response.function_call_arguments.done",
-            item_id="fc_123",
-            output_index=0,
-            name="get_weather",
-            arguments='{"city": "Paris"}',
-        ),
-        types.SimpleNamespace(
-            type="response.output_item.done",
-            output_index=0,
-            item=types.SimpleNamespace(
-                type="function_call",
-                id="fc_123",
-                call_id="call_123",
-                name="get_weather",
-                arguments='{"city": "Paris"}',
-            ),
-        ),
-        types.SimpleNamespace(
-            type="response.completed",
-            response=types.SimpleNamespace(
-                id="resp_123",
-                status="completed",
-                usage=None,
-            ),
-        ),
-    ]
-
-    class MockAsyncStream:
-        def __init__(self, events: list[Any]) -> None:
-            self._events = events
-            self._index = 0
-
-        def __aiter__(self) -> "MockAsyncStream":
-            return self
-
-        async def __anext__(self) -> Any:
-            if self._index >= len(self._events):
-                raise StopAsyncIteration
-            event = self._events[self._index]
-            self._index += 1
-            return event
-
-    async def mock_create(**_: Any) -> MockAsyncStream:
-        return MockAsyncStream(stream_events)
-
-    fake_client = types.SimpleNamespace(
-        responses=types.SimpleNamespace(create=mock_create)
-    )
-
-    with patch.object(llm, "_get_async_client", return_value=fake_client):
-        with patch("crewai.events.event_bus.CrewAIEventsBus.emit") as mock_emit:
-            await llm._ahandle_streaming_responses({"input": []})
-
-    tool_call_events = [
-        call.kwargs["event"]
-        for call in mock_emit.call_args_list
-        if isinstance(call.kwargs.get("event"), LLMStreamChunkEvent)
-        and call.kwargs["event"].call_type == LLMCallType.TOOL_CALL
-    ]
-
-    assert [event.chunk for event in tool_call_events] == [
-        "",
-        '{"city"',
-        ': "Paris"}',
-    ]
-    assert [
-        event.tool_call.function.arguments for event in tool_call_events
-    ] == ["", '{"city"', '{"city": "Paris"}']
-    assert all(event.tool_call.id == "call_123" for event in tool_call_events)
-    assert all(
-        event.tool_call.function.name == "get_weather" for event in tool_call_events
-    )
-
-
 @pytest.mark.asyncio
 async def test_openai_async_streaming_returns_tool_calls_without_available_functions():
     """Test that async streaming returns tool calls list when available_functions is None.

lib/crewai/tests/project/test_json_loader.py

@@ -355,6 +355,24 @@ class TestLoadAgent:
         with pytest.raises(Exception):
             load_agent(agent_file)
 
+    @pytest.mark.parametrize("field", ["role", "goal", "backstory"])
+    def test_load_agent_rejects_null_required_fields(
+        self, tmp_path: Path, field: str
+    ):
+        agent_def = {
+            "role": "Researcher",
+            "goal": "Find information",
+            "backstory": "Expert researcher.",
+        }
+        agent_def[field] = None
+        agent_file = tmp_path / "agent.json"
+        agent_file.write_text(json.dumps(agent_def))
+
+        with pytest.raises(
+            JSONProjectValidationError, match=f"missing required field '{field}'"
+        ):
+            load_agent(agent_file)
+
     def test_load_agent_file_not_found(self):
         with pytest.raises(FileNotFoundError):
             load_agent(Path("/nonexistent/agent.json"))

lib/crewai/tests/test_flow_from_definition.py

@@ -1163,6 +1163,139 @@ methods:
     }
 
 
+def test_agent_action_runs_repository_yaml_definition(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Agent
+    from crewai.plus_api import PlusAPI
+
+    fetched_agents: list[str] = []
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository specialist",
+                "goal": "Answer support questions",
+                "backstory": "Loaded from the agent repository.",
+                "max_iter": 3,
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        fetched_agents.append(handle)
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Agent, messages: str, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {"agent": self.role, "input": messages, "max_iter": self.max_iter}
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: AgentFlow
+methods:
+  answer:
+    do:
+      call: agent
+      with:
+        from_repository: support_specialist
+        input: "${state.question}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    assert flow.kickoff(inputs={"question": "What is CrewAI?"}) == {
+        "agent": "Repository specialist",
+        "input": "What is CrewAI?",
+        "max_iter": 3,
+    }
+    assert fetched_agents == ["support_specialist"]
+
+
+def test_agent_action_repository_fetch_does_not_block_event_loop(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Agent
+    from crewai.plus_api import PlusAPI
+
+    loop_marker_ran = threading.Event()
+    fetch_started = threading.Event()
+    release_fetch = threading.Event()
+    fetch_saw_loop_marker = False
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository specialist",
+                "goal": "Answer support questions",
+                "backstory": "Loaded from the agent repository.",
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        nonlocal fetch_saw_loop_marker
+        fetch_started.set()
+        release_fetch.wait(timeout=1)
+        fetch_saw_loop_marker = loop_marker_ran.is_set()
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Agent, messages: str, **_kwargs: Any
+    ) -> str:
+        return f"{self.role}:{messages}"
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: AgentFlow
+methods:
+  answer:
+    do:
+      call: agent
+      with:
+        from_repository: support_specialist
+        input: "${state.question}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    async def run_flow() -> str:
+        async def mark_loop_progress() -> None:
+            while not fetch_started.is_set():
+                await asyncio.sleep(0)
+            loop_marker_ran.set()
+            release_fetch.set()
+
+        marker_task = asyncio.create_task(mark_loop_progress())
+        kickoff_task = asyncio.create_task(
+            flow.kickoff_async(inputs={"question": "What is CrewAI?"})
+        )
+        try:
+            result = await asyncio.wait_for(kickoff_task, timeout=2)
+            await asyncio.wait_for(marker_task, timeout=2)
+            return result
+        finally:
+            release_fetch.set()
+
+    assert asyncio.run(run_flow()) == "Repository specialist:What is CrewAI?"
+    assert fetch_saw_loop_marker
+
+
 def test_agent_action_renders_text_custom_expression_input(
     monkeypatch: pytest.MonkeyPatch,
 ):
@@ -1281,6 +1414,7 @@ def test_agent_action_json_schema_describes_inline_agent_definitions():
         "role",
         "goal",
         "backstory",
+        "from_repository",
         "settings",
         "llm",
         "input",
@@ -1385,6 +1519,167 @@ methods:
     }
 
 
+def test_crew_action_runs_repository_agent_yaml_definition(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Crew
+    from crewai.plus_api import PlusAPI
+
+    fetched_agents: list[str] = []
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository researcher",
+                "goal": "Research {topic}",
+                "backstory": "Loaded from the agent repository.",
+                "max_iter": 5,
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        fetched_agents.append(handle)
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Crew, inputs: dict[str, Any] | None = None, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {
+            "crew": self.name,
+            "agents": [
+                {"role": agent.role, "max_iter": agent.max_iter}
+                for agent in self.agents
+            ],
+            "tasks": [task.description for task in self.tasks],
+            "inputs": inputs,
+        }
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Crew, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: CrewFlow
+methods:
+  research:
+    do:
+      call: crew
+      with:
+        name: inline_research
+        agents:
+          researcher:
+            from_repository: researcher
+        tasks:
+          - name: research_task
+            description: Research {topic}
+            expected_output: Findings about {topic}
+            agent: researcher
+      inputs:
+        topic: "${state.topic}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    assert flow.kickoff(inputs={"topic": "AI"}) == {
+        "crew": "inline_research",
+        "agents": [{"role": "Repository researcher", "max_iter": 5}],
+        "tasks": ["Research {topic}"],
+        "inputs": {"topic": "AI"},
+    }
+    assert fetched_agents == ["researcher"]
+
+
+def test_crew_action_repository_fetch_does_not_block_event_loop(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    from crewai import Crew
+    from crewai.plus_api import PlusAPI
+
+    loop_marker_ran = threading.Event()
+    fetch_started = threading.Event()
+    release_fetch = threading.Event()
+    fetch_saw_loop_marker = False
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def json(self) -> dict[str, Any]:
+            return {
+                "role": "Repository researcher",
+                "goal": "Research {topic}",
+                "backstory": "Loaded from the agent repository.",
+                "tools": [],
+            }
+
+    def fake_get_agent(self: PlusAPI, handle: str) -> FakeResponse:
+        nonlocal fetch_saw_loop_marker
+        fetch_started.set()
+        release_fetch.wait(timeout=1)
+        fetch_saw_loop_marker = loop_marker_ran.is_set()
+        return FakeResponse()
+
+    async def fake_kickoff_async(
+        self: Crew, inputs: dict[str, Any] | None = None, **_kwargs: Any
+    ) -> dict[str, Any]:
+        return {"agents": [agent.role for agent in self.agents], "inputs": inputs}
+
+    monkeypatch.setattr("crewai.auth.token.get_auth_token", lambda: "test-token")
+    monkeypatch.setattr(PlusAPI, "get_agent", fake_get_agent)
+    monkeypatch.setattr(Crew, "kickoff_async", fake_kickoff_async)
+
+    yaml_str = """
+schema: crewai.flow/v1
+name: CrewFlow
+methods:
+  research:
+    do:
+      call: crew
+      with:
+        agents:
+          researcher:
+            from_repository: researcher
+        tasks:
+          - description: Research {topic}
+            expected_output: Findings about {topic}
+            agent: researcher
+      inputs:
+        topic: "${state.topic}"
+    start: true
+"""
+
+    flow = Flow.from_declaration(contents=yaml_str)
+
+    async def run_flow() -> dict[str, Any]:
+        async def mark_loop_progress() -> None:
+            while not fetch_started.is_set():
+                await asyncio.sleep(0)
+            loop_marker_ran.set()
+            release_fetch.set()
+
+        marker_task = asyncio.create_task(mark_loop_progress())
+        kickoff_task = asyncio.create_task(
+            flow.kickoff_async(inputs={"topic": "AI"})
+        )
+        try:
+            result = await asyncio.wait_for(kickoff_task, timeout=2)
+            await asyncio.wait_for(marker_task, timeout=2)
+            return result
+        finally:
+            release_fetch.set()
+
+    assert asyncio.run(run_flow()) == {
+        "agents": ["Repository researcher"],
+        "inputs": {"topic": "AI"},
+    }
+    assert fetch_saw_loop_marker
+
+
 def test_crew_action_interpolates_runtime_strings_and_lists(
     monkeypatch: pytest.MonkeyPatch,
 ):
@@ -1709,6 +2004,7 @@ def test_crew_action_json_schema_describes_inline_crew_definitions():
         "role",
         "goal",
         "backstory",
+        "from_repository",
         "settings",
         "llm",
         "tools",
@@ -1728,36 +2024,45 @@ def test_crew_action_json_schema_describes_inline_crew_definitions():
 
 
 def test_crew_action_rejects_incomplete_inline_agent_definition():
-    with pytest.raises(ValidationError, match="goal"):
-        FlowDefinition.from_declaration(contents=
-            {
-                "schema": "crewai.flow/v1",
-                "name": "CrewFlow",
-                "methods": {
-                    "research": {
-                        "start": True,
-                        "do": {
-                            "call": "crew",
-                            "with": {
-                                "agents": {
-                                    "researcher": {
-                                        "role": "Researcher",
-                                        "backstory": "Knows things.",
-                                    }
-                                },
-                                "tasks": [
-                                    {
-                                        "description": "Research",
-                                        "expected_output": "Findings",
-                                        "agent": "researcher",
-                                    }
-                                ],
+    from crewai.project.crew_loader import load_crew_from_definition
+    from crewai.project.json_loader import JSONProjectValidationError
+
+    definition = FlowDefinition.from_declaration(contents=
+        {
+            "schema": "crewai.flow/v1",
+            "name": "CrewFlow",
+            "methods": {
+                "research": {
+                    "start": True,
+                    "do": {
+                        "call": "crew",
+                        "with": {
+                            "agents": {
+                                "researcher": {
+                                    "role": "Researcher",
+                                    "backstory": "Knows things.",
+                                }
                             },
+                            "tasks": [
+                                {
+                                    "description": "Research",
+                                    "expected_output": "Findings",
+                                    "agent": "researcher",
+                                }
+                            ],
                         },
-                    }
-                },
-            }
-        )
+                    },
+                }
+            },
+        }
+    )
+    crew_definition = definition.methods["research"].do.with_
+    assert crew_definition.agents["researcher"].goal is None
+
+    with pytest.raises(
+        JSONProjectValidationError, match="missing required field 'goal'"
+    ):
+        load_crew_from_definition(crew_definition, source="crew action")
 
 
 def test_crew_action_rejects_python_ref_field():

lib/crewai/tests/test_task.py

@@ -931,6 +931,48 @@ def test_interpolate_inputs(tmp_path):
     assert task.output_file == str(tmp_path / "ML" / "output_2025.txt")
 
 
+@pytest.mark.parametrize(
+    ("template", "malicious_inputs", "expected_error"),
+    [
+        ("reports/{name}.md", {"name": "../../../../tmp/pwn"}, "path traversal"),
+        ("{p}", {"p": "/tmp/abs_pwn"}, "absolute paths"),
+        ("{p}", {"p": "~/.bashrc"}, "shell expansion"),
+        ("{p}", {"p": "x;rm -rf /"}, "shell special characters"),
+        ("{p}", {"p": r"C:\Windows\evil"}, "absolute paths"),
+    ],
+)
+def test_interpolate_output_file_rejects_unsafe_inputs(
+    template, malicious_inputs, expected_error
+):
+    """Untrusted inputs must not escape the output_file path via interpolation."""
+    task = Task(
+        description="do {p} {name}".replace("{p}", "x").replace("{name}", "x"),
+        expected_output="e",
+        output_file=template,
+    )
+    with pytest.raises(ValueError, match=expected_error):
+        task.interpolate_inputs_and_add_conversation_history(inputs=malicious_inputs)
+
+
+def test_interpolate_output_file_allows_safe_inputs(tmp_path):
+    """Safe input values and developer-chosen absolute base paths still work."""
+    task = Task(
+        description="d",
+        expected_output="e",
+        output_file="reports/{name}.md",
+    )
+    task.interpolate_inputs_and_add_conversation_history(inputs={"name": "q3_summary"})
+    assert task.output_file == "reports/q3_summary.md"
+
+    abs_task = Task(
+        description="d",
+        expected_output="e",
+        output_file=str(tmp_path / "{topic}" / "out.md"),
+    )
+    abs_task.interpolate_inputs_and_add_conversation_history(inputs={"topic": "sales"})
+    assert abs_task.output_file == str(tmp_path / "sales" / "out.md")
+
+
 def test_interpolate_only():
     """Test the interpolate_only method for various scenarios including JSON structure preservation."""
 

lib/crewai/tests/utilities/test_console_formatter_pause_resume.py

@@ -1,6 +1,5 @@
 from unittest.mock import MagicMock, patch
 from rich.live import Live
-from crewai.events.types.llm_events import LLMCallType
 from crewai.events.utils.console_formatter import ConsoleFormatter
 
 
@@ -111,38 +110,3 @@ class TestConsoleFormatterPauseResume:
             # Streaming again creates new session
             formatter.handle_llm_stream_chunk("chunk 2", call_type=None)
             assert formatter._streaming_live == mock_live_instance_2
-
-    def test_tool_call_stream_chunk_does_not_create_live_panel(self):
-        formatter = ConsoleFormatter(verbose=True)
-
-        with patch("crewai.events.utils.console_formatter.Live") as mock_live_class:
-            formatter.handle_llm_stream_chunk(
-                '{"city":"Paris"}', call_type=LLMCallType.TOOL_CALL
-            )
-
-        mock_live_class.assert_not_called()
-        assert formatter._streaming_live is None
-        assert formatter._is_streaming is False
-
-    def test_tool_call_stream_chunk_stops_existing_live_panel(self):
-        formatter = ConsoleFormatter(verbose=True)
-        mock_live = MagicMock(spec=Live)
-        formatter._streaming_live = mock_live
-
-        formatter.handle_llm_stream_chunk(
-            '{"city":"Paris"}', call_type=LLMCallType.TOOL_CALL
-        )
-
-        mock_live.stop.assert_called_once()
-        assert formatter._streaming_live is None
-
-    def test_tool_usage_started_pauses_existing_live_panel(self):
-        formatter = ConsoleFormatter(verbose=True)
-        mock_live = MagicMock(spec=Live)
-        formatter._streaming_live = mock_live
-
-        with patch.object(formatter, "print_panel"):
-            formatter.handle_tool_usage_started("get_weather", {"city": "Paris"})
-
-        mock_live.stop.assert_called_once()
-        assert formatter._streaming_live is None