Fix type validation error in hierarchical process delegation (Issue #2606)

src/crewai/tools/agent_tools/delegate_work_tool.py

@@ -1,13 +1,14 @@
-from typing import Optional
+import json
+from typing import Any, Dict, Optional, Union
 
 from pydantic import BaseModel, Field
 
 from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
 
 
 class DelegateWorkToolSchema(BaseModel):
-    task: str = Field(..., description="The task to delegate")
-    context: str = Field(..., description="The context for the task")
+    task: Union[str, Dict[str, Any]] = Field(..., description="The task to delegate")
+    context: Union[str, Dict[str, Any]] = Field(..., description="The context for the task")
     coworker: str = Field(
         ..., description="The role/name of the coworker to delegate to"
     )
@@ -21,10 +22,12 @@ class DelegateWorkTool(BaseAgentTool):
 
     def _run(
         self,
-        task: str,
-        context: str,
+        task: Union[str, Dict[str, Any]],
+        context: Union[str, Dict[str, Any]],
         coworker: Optional[str] = None,
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
-        return self._execute(coworker, task, context)
+        task_str = json.dumps(task) if isinstance(task, dict) else task
+        context_str = json.dumps(context) if isinstance(context, dict) else context
+        return self._execute(coworker, task_str, context_str)

tests/tools/agent_tools/test_delegate_work_tool_dict_input.py

@@ -0,0 +1,37 @@
+"""Test delegate work tool with dictionary inputs."""
+
+import pytest
+
+from crewai.agent import Agent
+from crewai.tools.agent_tools.agent_tools import AgentTools
+
+researcher = Agent(
+    role="researcher",
+    goal="make the best research and analysis on content about AI and AI agents",
+    backstory="You're an expert researcher, specialized in technology",
+    allow_delegation=False,
+)
+tools = AgentTools(agents=[researcher]).tools()
+delegate_tool = tools[0]
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_delegate_work_with_dict_input():
+    """Test that the delegate work tool can handle dictionary inputs."""
+    task_dict = {
+        "description": "share your take on AI Agents",
+        "goal": "provide comprehensive analysis"
+    }
+    context_dict = {
+        "background": "I heard you hate them",
+        "additional_info": "We need this for a report"
+    }
+
+    result = delegate_tool.run(
+        coworker="researcher",
+        task=task_dict,
+        context=context_dict,
+    )
+
+    assert isinstance(result, str)
+    assert len(result) > 0