Compare commits

...

4 Commits

Author SHA1 Message Date
Devin AI
d372b3d9da fix: remove duplicate constraint injection in context, fix unused import
- Remove constraint text injection into context string (constraints are
  already rendered by Task.prompt() via the constraints field)
- Remove unused MagicMock import from test file
- Update tests to verify context is passed through unchanged

Addresses review feedback from Cursor Bugbot and github-code-quality bot.

Co-Authored-By: João <joao@crewai.com>
2026-04-15 20:10:08 +00:00
Devin AI
294e97277d fix: update BaseAgent.get_delegation_tools signature to accept task parameter
Co-Authored-By: João <joao@crewai.com>
2026-04-15 19:46:32 +00:00
Devin AI
9fac35bb05 style: apply ruff format to modified files
Co-Authored-By: João <joao@crewai.com>
2026-04-15 19:41:51 +00:00
Devin AI
fa67887ac2 fix: preserve task constraints during multi-agent delegation (#5476)
When agents delegate work to other agents, structured constraints
(domain scope, quality requirements, temporal/geographic limits) were
silently lost because only the natural language task description was
passed to the delegated task.

Changes:
- Add 'constraints' field to Task model (list[str], default empty)
- Include constraints in Task.prompt() output when present
- Add 'original_task' field to BaseAgentTool to track the source task
- Propagate constraints from original task to delegated task in
  BaseAgentTool._execute() - both as structured field and in context
- Update AgentTools, Agent.get_delegation_tools(), and Crew delegation
  methods to thread the task through the delegation chain
- Update OpenAI and LangGraph adapter get_delegation_tools signatures
- Add info logging when constraints are propagated
- All changes are backward compatible (task parameter is optional)

Co-Authored-By: João <joao@crewai.com>
2026-04-15 19:38:19 +00:00
9 changed files with 439 additions and 14 deletions

View File

@@ -1091,8 +1091,10 @@ class Agent(BaseAgent):
)
)
def get_delegation_tools(self, agents: Sequence[BaseAgent]) -> list[BaseTool]:
agent_tools = AgentTools(agents=agents)
def get_delegation_tools(
self, agents: Sequence[BaseAgent], task: Task | None = None
) -> list[BaseTool]:
agent_tools = AgentTools(agents=agents, task=task)
return agent_tools.tools()
def get_platform_tools(self, apps: list[PlatformAppOrAction]) -> list[BaseTool]:

View File

@@ -274,18 +274,22 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
available_tools: list[Any] = self._tool_adapter.tools()
self._graph.tools = available_tools
def get_delegation_tools(self, agents: Sequence[BaseAgent]) -> list[BaseTool]:
def get_delegation_tools(
self, agents: Sequence[BaseAgent], task: Any | None = None
) -> list[BaseTool]:
"""Implement delegation tools support for LangGraph.
Creates delegation tools that allow this agent to delegate tasks to other agents.
When a task is provided, its constraints are propagated to the delegation tools.
Args:
agents: List of agents available for delegation.
task: Optional task whose constraints should be propagated.
Returns:
List of delegation tools.
"""
agent_tools: AgentTools = AgentTools(agents=agents)
agent_tools: AgentTools = AgentTools(agents=agents, task=task)
return agent_tools.tools()
@staticmethod

View File

@@ -223,18 +223,22 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
"""
return self._converter_adapter.post_process_result(result.final_output)
def get_delegation_tools(self, agents: Sequence[BaseAgent]) -> list[BaseTool]:
def get_delegation_tools(
self, agents: Sequence[BaseAgent], task: Any | None = None
) -> list[BaseTool]:
"""Implement delegation tools support.
Creates delegation tools that allow this agent to delegate tasks to other agents.
When a task is provided, its constraints are propagated to the delegation tools.
Args:
agents: List of agents available for delegation.
task: Optional task whose constraints should be propagated.
Returns:
List of delegation tools.
"""
agent_tools: AgentTools = AgentTools(agents=agents)
agent_tools: AgentTools = AgentTools(agents=agents, task=task)
return agent_tools.tools()
def configure_structured_output(self, task: Any) -> None:

View File

@@ -530,7 +530,9 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
pass
@abstractmethod
def get_delegation_tools(self, agents: Sequence[BaseAgent]) -> list[BaseTool]:
def get_delegation_tools(
self, agents: Sequence[BaseAgent], task: Any | None = None
) -> list[BaseTool]:
"""Set the task tools that init BaseAgenTools class."""
@abstractmethod

View File

@@ -1608,9 +1608,10 @@ class Crew(FlowTrackable, BaseModel):
tools: list[BaseTool],
task_agent: BaseAgent,
agents: Sequence[BaseAgent],
task: Task | None = None,
) -> list[BaseTool]:
if hasattr(task_agent, "get_delegation_tools"):
delegation_tools = task_agent.get_delegation_tools(agents)
delegation_tools = task_agent.get_delegation_tools(agents, task=task)
# Cast delegation_tools to the expected type for _merge_tools
return self._merge_tools(tools, delegation_tools)
return tools
@@ -1693,7 +1694,7 @@ class Crew(FlowTrackable, BaseModel):
if not tools:
tools = []
tools = self._inject_delegation_tools(
tools, task.agent, agents_for_delegation
tools, task.agent, agents_for_delegation, task=task
)
return tools
@@ -1723,10 +1724,12 @@ class Crew(FlowTrackable, BaseModel):
) -> list[BaseTool]:
if self.manager_agent:
if task.agent:
tools = self._inject_delegation_tools(tools, task.agent, [task.agent])
tools = self._inject_delegation_tools(
tools, task.agent, [task.agent], task=task
)
else:
tools = self._inject_delegation_tools(
tools, self.manager_agent, self.agents
tools, self.manager_agent, self.agents, task=task
)
return tools

View File

@@ -193,6 +193,13 @@ class Task(BaseModel):
description="A converter class used to export structured output",
default=None,
)
constraints: list[str] = Field(
default_factory=list,
description="Structured constraints that must be preserved during task delegation. "
"Each constraint is a string describing a specific requirement (e.g., domain scope, "
"quality specs, temporal or geographic limits). These are automatically propagated "
"to delegated tasks so worker agents are aware of all original constraints.",
)
processed_by_agents: set[str] = Field(default_factory=set)
guardrail: GuardrailType | None = Field(
default=None,
@@ -901,10 +908,17 @@ class Task(BaseModel):
tasks_slices = [description]
if self.constraints:
constraints_text = (
"\n\nTask Constraints (MUST be respected):\n"
+ "\n".join(f"- {constraint}" for constraint in self.constraints)
)
tasks_slices.append(constraints_text)
output = I18N_DEFAULT.slice("expected_output").format(
expected_output=self.expected_output
)
tasks_slices = [description, output]
tasks_slices.append(output)
if self.markdown:
markdown_instruction = """Your final answer MUST be formatted in Markdown syntax.

View File

@@ -10,26 +10,34 @@ from crewai.utilities.i18n import I18N_DEFAULT
if TYPE_CHECKING:
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.task import Task
from crewai.tools.base_tool import BaseTool
class AgentTools:
"""Manager class for agent-related tools"""
def __init__(self, agents: Sequence[BaseAgent]) -> None:
def __init__(self, agents: Sequence[BaseAgent], task: Task | None = None) -> None:
self.agents = agents
self.task = task
def tools(self) -> list[BaseTool]:
"""Get all available agent tools"""
"""Get all available agent tools.
When a task is provided, its constraints are automatically propagated
to the delegation tools so that worker agents receive them.
"""
coworkers = ", ".join([f"{agent.role}" for agent in self.agents])
delegate_tool = DelegateWorkTool(
agents=self.agents,
original_task=self.task,
description=I18N_DEFAULT.tools("delegate_work").format(coworkers=coworkers), # type: ignore
)
ask_tool = AskQuestionTool(
agents=self.agents,
original_task=self.task,
description=I18N_DEFAULT.tools("ask_question").format(coworkers=coworkers), # type: ignore
)

View File

@@ -16,6 +16,10 @@ class BaseAgentTool(BaseTool):
"""Base class for agent-related tools"""
agents: list[BaseAgent] = Field(description="List of available agents")
original_task: Task | None = Field(
default=None,
description="The original task being delegated, used to propagate constraints",
)
def sanitize_agent_name(self, name: str) -> str:
"""
@@ -51,6 +55,10 @@ class BaseAgentTool(BaseTool):
"""
Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
When the original_task has constraints defined, they are automatically
propagated to the delegated Task object. The constraints are then
rendered by Task.prompt() so the worker agent sees them.
Args:
agent_name: Name/role of the agent to delegate to (case-insensitive)
task: The specific question or task to delegate
@@ -114,10 +122,25 @@ class BaseAgentTool(BaseTool):
selected_agent = agent[0]
try:
# Propagate constraints from the original task to the delegated task.
# Constraints are set on the Task object so that Task.prompt() renders
# them for the worker agent — no need to also inject them into `context`,
# which would cause duplication.
constraints: list[str] = []
if self.original_task and self.original_task.constraints:
constraints = list(self.original_task.constraints)
logger.info(
"Propagating %d constraint(s) from original task to delegated task for agent '%s': %s",
len(constraints),
self.sanitize_agent_name(selected_agent.role),
constraints,
)
task_with_assigned_agent = Task(
description=task,
agent=selected_agent,
expected_output=I18N_DEFAULT.slice("manager_request"),
constraints=constraints,
)
logger.debug(
f"Created task for agent '{self.sanitize_agent_name(selected_agent.role)}': {task}"

View File

@@ -0,0 +1,365 @@
"""Tests for constraint propagation during task delegation.
These tests verify that when a Task has structured constraints defined,
they are properly propagated to delegated tasks through the DelegateWorkTool
and AskQuestionTool, ensuring worker agents receive the original requirements.
See: https://github.com/crewAIInc/crewAI/issues/5476
"""
import logging
from unittest.mock import patch
import pytest
from crewai.agent import Agent
from crewai.task import Task
from crewai.tools.agent_tools.agent_tools import AgentTools
from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
from crewai.tools.agent_tools.delegate_work_tool import DelegateWorkTool
from crewai.tools.agent_tools.ask_question_tool import AskQuestionTool
@pytest.fixture
def researcher():
return Agent(
role="researcher",
goal="Research AI topics",
backstory="Expert researcher in AI",
allow_delegation=False,
)
@pytest.fixture
def writer():
return Agent(
role="writer",
goal="Write articles about AI",
backstory="Expert technical writer",
allow_delegation=False,
)
@pytest.fixture
def task_with_constraints(researcher):
return Task(
description="Find the best open-source ML frameworks from 2024 in Europe",
expected_output="A list of ML frameworks",
agent=researcher,
constraints=[
"Only open-source frameworks",
"Must be from 2024",
"Only frameworks available in Europe",
],
)
@pytest.fixture
def task_without_constraints(researcher):
return Task(
description="Find ML frameworks",
expected_output="A list of ML frameworks",
agent=researcher,
)
class TestTaskConstraintsField:
"""Tests for the constraints field on the Task model."""
def test_task_has_constraints_field(self):
"""A Task can be created with a constraints field."""
task = Task(
description="Test task",
expected_output="Test output",
constraints=["constraint1", "constraint2"],
)
assert task.constraints == ["constraint1", "constraint2"]
def test_task_constraints_default_empty(self):
"""A Task without constraints has an empty list by default."""
task = Task(
description="Test task",
expected_output="Test output",
)
assert task.constraints == []
def test_task_prompt_includes_constraints(self):
"""Task.prompt() includes constraints when they are set."""
task = Task(
description="Find ML frameworks",
expected_output="A list of frameworks",
constraints=["Only open-source", "From 2024 only"],
)
prompt = task.prompt()
assert "Task Constraints (MUST be respected):" in prompt
assert "- Only open-source" in prompt
assert "- From 2024 only" in prompt
def test_task_prompt_excludes_constraints_when_empty(self):
"""Task.prompt() does not include constraint section when constraints are empty."""
task = Task(
description="Find ML frameworks",
expected_output="A list of frameworks",
)
prompt = task.prompt()
assert "Task Constraints" not in prompt
class TestConstraintPropagationInDelegation:
"""Tests for constraint propagation through delegation tools."""
def test_delegate_tool_receives_original_task(self, researcher, writer, task_with_constraints):
"""DelegateWorkTool is initialized with the original task reference."""
tools = AgentTools(agents=[writer], task=task_with_constraints).tools()
delegate_tool = tools[0]
assert isinstance(delegate_tool, DelegateWorkTool)
assert delegate_tool.original_task is task_with_constraints
def test_ask_tool_receives_original_task(self, researcher, writer, task_with_constraints):
"""AskQuestionTool is initialized with the original task reference."""
tools = AgentTools(agents=[writer], task=task_with_constraints).tools()
ask_tool = tools[1]
assert isinstance(ask_tool, AskQuestionTool)
assert ask_tool.original_task is task_with_constraints
def test_delegate_tool_without_task_has_none(self, writer):
"""When no task is provided, original_task is None."""
tools = AgentTools(agents=[writer]).tools()
delegate_tool = tools[0]
assert delegate_tool.original_task is None
@patch.object(Agent, "execute_task")
def test_constraints_propagated_to_delegated_task(
self, mock_execute, researcher, writer, task_with_constraints
):
"""Constraints from the original task are propagated to the delegated task."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_with_constraints).tools()
delegate_tool = tools[0]
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Need a comprehensive list",
)
# Verify execute_task was called
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
# The delegated task should have the constraints from the original task
assert delegated_task.constraints == [
"Only open-source frameworks",
"Must be from 2024",
"Only frameworks available in Europe",
]
# Context should NOT be modified — constraints are rendered via Task.prompt()
assert delegated_context == "Need a comprehensive list"
@patch.object(Agent, "execute_task")
def test_context_not_modified_by_constraints(
self, mock_execute, researcher, writer, task_with_constraints
):
"""Context is passed through unchanged; constraints live on the Task object."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_with_constraints).tools()
delegate_tool = tools[0]
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Previous context here",
)
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
# Context should be unchanged
assert delegated_context == "Previous context here"
# Constraints should be on the task object
assert len(delegated_task.constraints) == 3
@patch.object(Agent, "execute_task")
def test_no_constraints_no_modification(
self, mock_execute, researcher, writer, task_without_constraints
):
"""When original task has no constraints, context is not modified."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_without_constraints).tools()
delegate_tool = tools[0]
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Just context",
)
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
assert delegated_task.constraints == []
assert delegated_context == "Just context"
@patch.object(Agent, "execute_task")
def test_ask_question_propagates_constraints(
self, mock_execute, researcher, writer, task_with_constraints
):
"""AskQuestionTool also propagates constraints to the delegated task."""
mock_execute.return_value = "answer"
tools = AgentTools(agents=[researcher], task=task_with_constraints).tools()
ask_tool = tools[1]
ask_tool.run(
coworker="researcher",
question="What are the best frameworks?",
context="Need details",
)
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
assert delegated_task.constraints == task_with_constraints.constraints
# Context should be unchanged — constraints live on the task
assert delegated_context == "Need details"
@patch.object(Agent, "execute_task")
def test_constraints_propagated_when_no_original_context(
self, mock_execute, researcher, writer, task_with_constraints
):
"""Even with empty context, constraints are on the task, not injected into context."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_with_constraints).tools()
delegate_tool = tools[0]
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="",
)
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
# Context should remain empty
assert delegated_context == ""
# Constraints are on the task object
assert delegated_task.constraints == task_with_constraints.constraints
@patch.object(Agent, "execute_task")
def test_delegation_without_original_task_works(
self, mock_execute, researcher, writer
):
"""Delegation still works when no original task is set (backward compatible)."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher]).tools()
delegate_tool = tools[0]
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Some context",
)
mock_execute.assert_called_once()
delegated_task = mock_execute.call_args[0][0]
delegated_context = mock_execute.call_args[0][1]
# Should work normally without constraints
assert delegated_task.constraints == []
assert delegated_context == "Some context"
class TestConstraintPropagationLogging:
"""Tests for logging during constraint propagation."""
@patch.object(Agent, "execute_task")
def test_constraint_propagation_logs_info(
self, mock_execute, researcher, writer, task_with_constraints, caplog
):
"""An info log is emitted when constraints are propagated."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_with_constraints).tools()
delegate_tool = tools[0]
with caplog.at_level(logging.INFO, logger="crewai.tools.agent_tools.base_agent_tools"):
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Context",
)
assert any("Propagating 3 constraint(s)" in record.message for record in caplog.records)
@patch.object(Agent, "execute_task")
def test_no_log_when_no_constraints(
self, mock_execute, researcher, writer, task_without_constraints, caplog
):
"""No constraint propagation log when there are no constraints."""
mock_execute.return_value = "result"
tools = AgentTools(agents=[researcher], task=task_without_constraints).tools()
delegate_tool = tools[0]
with caplog.at_level(logging.INFO, logger="crewai.tools.agent_tools.base_agent_tools"):
delegate_tool.run(
coworker="researcher",
task="Find ML frameworks",
context="Context",
)
assert not any("Propagating" in record.message for record in caplog.records)
class TestAgentToolsTaskPassThrough:
"""Tests that AgentTools passes the task to the underlying tools."""
def test_agent_tools_with_task(self, researcher, task_with_constraints):
"""AgentTools passes the task to both delegate and ask tools."""
agent_tools = AgentTools(agents=[researcher], task=task_with_constraints)
tools = agent_tools.tools()
assert len(tools) == 2
for tool in tools:
assert isinstance(tool, BaseAgentTool)
assert tool.original_task is task_with_constraints
def test_agent_tools_without_task(self, researcher):
"""AgentTools without a task sets original_task to None on tools."""
agent_tools = AgentTools(agents=[researcher])
tools = agent_tools.tools()
assert len(tools) == 2
for tool in tools:
assert isinstance(tool, BaseAgentTool)
assert tool.original_task is None
def test_agent_get_delegation_tools_passes_task(self, researcher, task_with_constraints):
"""Agent.get_delegation_tools passes the task through to AgentTools."""
tools = researcher.get_delegation_tools(agents=[researcher], task=task_with_constraints)
assert len(tools) == 2
for tool in tools:
assert isinstance(tool, BaseAgentTool)
assert tool.original_task is task_with_constraints
def test_agent_get_delegation_tools_without_task(self, researcher):
"""Agent.get_delegation_tools without task still works (backward compatible)."""
tools = researcher.get_delegation_tools(agents=[researcher])
assert len(tools) == 2
for tool in tools:
assert isinstance(tool, BaseAgentTool)
assert tool.original_task is None