Fix DelegateWorkTool Type Validation in Hierarchical Process

[HaohanTsao]

Related Issue

Fixes #2606

Problem Description

When using the hierarchical process in CrewAI, manager agents cannot delegate tasks due to type validation errors. The DelegateWorkToolSchema requires string values for task and context parameters, but in hierarchical delegation, the manager is passing dictionary objects with format {'description': 'Task details', 'type': 'str'}.

Error message:

Tool Usage Failed
Name: Delegate work to coworker
Error: Arguments validation failed: 2 validation errors for DelegateWorkToolSchema
task
  Input should be a valid string [type=string_type, input_value={'description': 'Research...thcare.', 'type': 'str'}, input_type=dict]
context
  Input should be a valid string [type=string_type, input_value={'description': 'This tas...ations.', 'type': 'str'}, input_type=dict]

Changes Made

1. Modified src/crewai/tools/agent_tools/delegate_work_tool.py to:

   • Accept both string and dictionary inputs in the DelegateWorkToolSchema
   • Add type conversion logic in the _run method to convert dictionary inputs to strings when possible

2. Added comprehensive tests in tests/tools/agent_tools/test_delegate_work_tool.py to verify:

   • Original string input functionality still works (backward compatibility)
   • Dictionary inputs are properly handled and converted
   • Mixed input types (string + dictionary) are handled correctly
   • Edge cases are handled properly
   • Schema validation works with all input types

# Updated imports
from typing import Optional, Union, Dict, Any

# Updated schema to accept both string and dictionary inputs
class DelegateWorkToolSchema(BaseModel):
    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")

# Updated _run method to handle both input types
def _run(
    self,
    task: Union[str, Dict[str, Any]],
    context: Union[str, Dict[str, Any]],
    coworker: Optional[str] = None,
    **kwargs,
) -> str:
    # Convert task to string if it's a dictionary
    if isinstance(task, dict) and "description" in task:
        task = task["description"]
    
    # Convert context to string if it's a dictionary
    if isinstance(context, dict) and "description" in context:
        context = context["description"]
        
    coworker = self._get_coworker(coworker, **kwargs)
    return self._execute(coworker, task, context)

Testing

1. Manually tested the fix by creating a hierarchical crew with a manager agent that delegates tasks, fixed the errors happend in issue.
2. Added comprehensive test suite in test_delegate_work_tool.py covering all input scenarios

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist

• My code follows the code style of this project
• My change is minimal and focused on fixing the specific issue
• I have added tests to cover my changes