fix: stop output_pydantic from leaking into tool-calling loop (#5472)

lib/crewai/src/crewai/agent/core.py

@@ -1063,11 +1063,7 @@ def create_agent_executor(
                 respect_context_window=self.respect_context_window,
                 request_within_rpm_limit=rpm_limit_fn,
                 callbacks=[TokenCalcHandler(self._token_process)],
-                response_model=(
-                    task.response_model or task.output_pydantic or task.output_json
-                )
-                if task
-                else None,
+                response_model=task.response_model if task else None,
             )
 
     def _update_executor_parameters(
@@ -1104,9 +1100,7 @@ def _update_executor_parameters(
         self.agent_executor.tools_names = get_tool_names(tools)
         self.agent_executor.tools_description = render_text_description_and_args(tools)
         self.agent_executor.response_model = (
-            (task.response_model or task.output_pydantic or task.output_json)
-            if task
-            else None
+            task.response_model if task else None
         )
 
         self.agent_executor.tools_handler = self.tools_handler

lib/crewai/src/crewai/agents/crew_agent_executor.py

@@ -502,7 +502,7 @@ def _invoke_loop_native_tools(self) -> AgentFinish:
                     available_functions=None,
                     from_task=self.task,
                     from_agent=self.agent,
-                    response_model=self.response_model,
+                    response_model=None,
                     executor_context=self,
                     verbose=self.agent.verbose,
                 )
@@ -1314,7 +1314,7 @@ async def _ainvoke_loop_native_tools(self) -> AgentFinish:
                     available_functions=None,
                     from_task=self.task,
                     from_agent=self.agent,
-                    response_model=self.response_model,
+                    response_model=None,
                     executor_context=self,
                     verbose=self.agent.verbose,
                 )

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -1319,7 +1319,7 @@ def call_llm_native_tools(
                 available_functions=None,
                 from_task=self.task,
                 from_agent=self.agent,
-                response_model=self.response_model,
+                response_model=None,
                 executor_context=self,
                 verbose=self.agent.verbose,
             )

lib/crewai/tests/agents/test_agent.py

@@ -2622,3 +2622,88 @@ def nested_direct_call() -> None:
 
         assert seen == [{"Original:", "Observation:"}]
         assert shared.stop == ["Original:"]
+
+
+class TestOutputPydanticDoesNotLeakIntoResponseModel:
+    """Regression tests for issue #5472.
+
+    output_pydantic / output_json should NOT be mapped onto
+    executor.response_model. Only explicit task.response_model
+    should use native structured output.
+    """
+
+    def test_output_pydantic_does_not_set_response_model(self):
+        """output_pydantic should stay in post-processing, not set response_model."""
+        from pydantic import BaseModel, Field
+
+        class MyOutput(BaseModel):
+            answer: str = Field(description="The answer")
+
+        agent = Agent(role="test", goal="test", backstory="test")
+        task = Task(
+            description="test",
+            expected_output="test",
+            output_pydantic=MyOutput,
+        )
+
+        agent.create_agent_executor(task=task)
+        assert agent.agent_executor.response_model is None
+
+    def test_output_json_does_not_set_response_model(self):
+        """output_json should stay in post-processing, not set response_model."""
+        from pydantic import BaseModel, Field
+
+        class MyOutput(BaseModel):
+            answer: str = Field(description="The answer")
+
+        agent = Agent(role="test", goal="test", backstory="test")
+        task = Task(
+            description="test",
+            expected_output="test",
+            output_json=MyOutput,
+        )
+
+        agent.create_agent_executor(task=task)
+        assert agent.agent_executor.response_model is None
+
+    def test_explicit_response_model_still_works(self):
+        """Explicit task.response_model should still propagate to the executor."""
+        from pydantic import BaseModel, Field
+
+        class MyOutput(BaseModel):
+            answer: str = Field(description="The answer")
+
+        agent = Agent(role="test", goal="test", backstory="test")
+        task = Task(
+            description="test",
+            expected_output="test",
+            response_model=MyOutput,
+        )
+
+        agent.create_agent_executor(task=task)
+        assert agent.agent_executor.response_model is MyOutput
+
+    def test_update_executor_parameters_respects_fix(self):
+        """_update_executor_parameters should apply the same mapping fix."""
+        from pydantic import BaseModel, Field
+
+        class MyOutput(BaseModel):
+            answer: str = Field(description="The answer")
+
+        agent = Agent(role="test", goal="test", backstory="test")
+        task_with_pydantic = Task(
+            description="test",
+            expected_output="test",
+            output_pydantic=MyOutput,
+        )
+        task_with_response_model = Task(
+            description="test",
+            expected_output="test",
+            response_model=MyOutput,
+        )
+
+        agent.create_agent_executor(task=task_with_pydantic)
+        assert agent.agent_executor.response_model is None
+
+        agent.create_agent_executor(task=task_with_response_model)
+        assert agent.agent_executor.response_model is MyOutput