Implement review suggestions: improved error handling, type validation, and edge case tests

Co-Authored-By: Joe Moura <joao@crewai.com>

docs/enterprise/features/agent-repository.mdx

@@ -1,70 +0,0 @@
----
-title: "Agent Repository"
-description: "Store and retrieve agents for your CrewAI projects"
----
-
-# Agent Repository
-
-The Agent Repository allows you to store, manage, and reuse agents across your CrewAI projects. This feature streamlines the development process by enabling you to configure agents once and use them in multiple projects.
-
-## How It Works
-
-When you create an agent in the CrewAI interface, it's stored in the Agent Repository. You can then initialize these agents in your code using the `from_repository` parameter.
-
-## Usage
-
-To use an agent from the repository in your CrewAI project, initialize it with the following code:
-
-```python
-from crewai import Agent
-
-# Initialize the agent with its role
-agent = Agent(from_repository="python-job-researcher")
-```
-
-### Creating a Crew with Repository Agents
-
-```python
-from crewai import Agent, Crew, Task
-
-agent = Agent(from_repository="python-job-researcher")
-
-job_search_task = Task(
-    description="Search for recent Python developer job listings online",
-    expected_output="Markdown list of 5 recent Python developer jobs with details.",
-    agent=agent,
-)
-
-crew = Crew(agents=[agent], tasks=[job_search_task], verbose=True)
-
-result = crew.kickoff()
-print(result)
-```
-
-## Important Notes
-
-- The `from_repository` value must match the agent's role in a URL-safe format.
-- If you change an agent's role after creation, you must update the `from_repository` value in your code accordingly, or you won't be able to find the agent anymore.
-- Make sure you have permission to use the agent as mentioned in the key points.
-
-## Agent Configuration
-
-When configuring an agent in the repository, you can specify:
-
-1. **Role** - The agent's primary function (e.g., "Python Job Researcher")
-2. **Goal** - What the agent aims to achieve (e.g., "Find Python developer job opportunities")
-3. **Backstory** - Context for the agent's behavior
-4. **Tools** - Available capabilities for the agent to use when performing tasks
-5. **Visibility Controls** - Who can access and use the agent
-
-## Managing Agents
-
-The Agent Repository interface provides functionality to:
-
-- View all available agents
-- Add new agents
-- Edit existing agents
-- Delete agents
-- View agent details including usage examples
-
-By leveraging the Agent Repository, you can build more modular and reusable AI workflows while maintaining a central location for managing your agents.

src/crewai/crew.py

@@ -52,7 +52,7 @@ from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.tools.base_tool import BaseTool, Tool
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities import I18N, FileHandler, Logger, RPMController
-from crewai.utilities.constants import NOT_SPECIFIED, TRAINING_DATA_FILE
+from crewai.utilities.constants import TRAINING_DATA_FILE
 from crewai.utilities.evaluators.crew_evaluator_handler import CrewEvaluator
 from crewai.utilities.evaluators.task_evaluator import TaskEvaluator
 from crewai.utilities.events.crew_events import (
@@ -478,7 +478,7 @@ class Crew(FlowTrackable, BaseModel):
         separated by a synchronous task.
         """
         for i, task in enumerate(self.tasks):
-            if task.async_execution and isinstance(task.context, list):
+            if task.async_execution and task.context:
                 for context_task in task.context:
                     if context_task.async_execution:
                         for j in range(i - 1, -1, -1):
@@ -496,7 +496,7 @@ class Crew(FlowTrackable, BaseModel):
         task_indices = {id(task): i for i, task in enumerate(self.tasks)}
 
         for task in self.tasks:
-            if isinstance(task.context, list):
+            if task.context:
                 for context_task in task.context:
                     if id(context_task) not in task_indices:
                         continue  # Skip context tasks not in the main tasks list
@@ -1034,14 +1034,11 @@ class Crew(FlowTrackable, BaseModel):
                 )
         return cast(List[BaseTool], tools)
 
-    def _get_context(self, task: Task, task_outputs: List[TaskOutput]) -> str:
-        if not task.context:
-            return ""
-
+    def _get_context(self, task: Task, task_outputs: List[TaskOutput]):
         context = (
-            aggregate_raw_outputs_from_task_outputs(task_outputs)
-            if task.context is NOT_SPECIFIED
-            else aggregate_raw_outputs_from_tasks(task.context)
+            aggregate_raw_outputs_from_tasks(task.context)
+            if task.context
+            else aggregate_raw_outputs_from_task_outputs(task_outputs)
         )
         return context
 
@@ -1229,7 +1226,7 @@ class Crew(FlowTrackable, BaseModel):
             task_mapping[task.key] = cloned_task
 
         for cloned_task, original_task in zip(cloned_tasks, self.tasks):
-            if isinstance(original_task.context, list):
+            if original_task.context:
                 cloned_context = [
                     task_mapping[context_task.key]
                     for context_task in original_task.context

src/crewai/task.py

@@ -2,6 +2,7 @@ import datetime
 import inspect
 import json
 import logging
+import re
 import threading
 import uuid
 from concurrent.futures import Future
@@ -40,7 +41,6 @@ from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
 from crewai.tools.base_tool import BaseTool
 from crewai.utilities.config import process_config
-from crewai.utilities.constants import NOT_SPECIFIED
 from crewai.utilities.converter import Converter, convert_to_model
 from crewai.utilities.events import (
     TaskCompletedEvent,
@@ -97,7 +97,7 @@ class Task(BaseModel):
     )
     context: Optional[List["Task"]] = Field(
         description="Other tasks that will have their output used as context for this task.",
-        default=NOT_SPECIFIED,
+        default=None,
     )
     async_execution: Optional[bool] = Field(
         description="Whether the task should be executed asynchronously or not.",
@@ -643,7 +643,7 @@ class Task(BaseModel):
 
         cloned_context = (
             [task_mapping[context_task.key] for context_task in self.context]
-            if isinstance(self.context, list)
+            if self.context
             else None
         )
 

src/crewai/telemetry/telemetry.py

@@ -10,18 +10,6 @@ from contextlib import contextmanager
 from importlib.metadata import version
 from typing import TYPE_CHECKING, Any, Optional
 
-from opentelemetry import trace
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
-    OTLPSpanExporter,
-)
-from opentelemetry.sdk.resources import SERVICE_NAME, Resource
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import (
-    BatchSpanProcessor,
-    SpanExportResult,
-)
-from opentelemetry.trace import Span, Status, StatusCode
-
 from crewai.telemetry.constants import (
     CREWAI_TELEMETRY_BASE_URL,
     CREWAI_TELEMETRY_SERVICE_NAME,
@@ -37,6 +25,18 @@ def suppress_warnings():
         yield
 
 
+from opentelemetry import trace  # noqa: E402
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
+    OTLPSpanExporter,  # noqa: E402
+)
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource  # noqa: E402
+from opentelemetry.sdk.trace import TracerProvider  # noqa: E402
+from opentelemetry.sdk.trace.export import (  # noqa: E402
+    BatchSpanProcessor,
+    SpanExportResult,
+)
+from opentelemetry.trace import Span, Status, StatusCode  # noqa: E402
+
 if TYPE_CHECKING:
     from crewai.crew import Crew
     from crewai.task import Task
@@ -232,7 +232,7 @@ class Telemetry:
                                 "agent_key": task.agent.key if task.agent else None,
                                 "context": (
                                     [task.description for task in task.context]
-                                    if isinstance(task.context, list)
+                                    if task.context
                                     else None
                                 ),
                                 "tools_names": [
@@ -748,7 +748,7 @@ class Telemetry:
                             "agent_key": task.agent.key if task.agent else None,
                             "context": (
                                 [task.description for task in task.context]
-                                if isinstance(task.context, list)
+                                if task.context
                                 else None
                             ),
                             "tools_names": [

src/crewai/tools/structured_tool.py

@@ -173,11 +173,18 @@ class CrewStructuredTool:
     def _parse_args(self, raw_args: Union[str, dict]) -> dict:
         """Parse and validate the input arguments against the schema.
 
+        This method handles different input formats from various LLM providers,
+        including nested dictionaries with 'value' fields that some providers use.
+
         Args:
-            raw_args: The raw arguments to parse, either as a string or dict
+            raw_args: The raw arguments to parse, either as a string or dict.
+                     Supports nested dictionaries with 'value' field for LLM provider compatibility.
 
         Returns:
             The validated arguments as a dictionary
+
+        Raises:
+            ValueError: If argument parsing or validation fails
         """
         if isinstance(raw_args, str):
             try:
@@ -187,6 +194,31 @@ class CrewStructuredTool:
             except json.JSONDecodeError as e:
                 raise ValueError(f"Failed to parse arguments as JSON: {e}")
 
+        # Handle nested dictionaries with 'value' field for all parameter types
+        if isinstance(raw_args, dict):
+            schema_fields = self.args_schema.model_fields
+            
+            for field_name, field_value in list(raw_args.items()):
+                # Check if this field exists in the schema
+                if field_name in schema_fields:
+                    # Handle nested dictionaries with 'value' field
+                    if isinstance(field_value, dict):
+                        if 'value' in field_value:
+                            # Extract the value from the nested dictionary
+                            value = field_value['value']
+                            self._logger.debug(f"Extracting value from nested dict for {field_name}")
+                            
+                            expected_type = schema_fields[field_name].annotation
+                            
+                            if expected_type in (str, int, float, bool) and not isinstance(value, expected_type):
+                                self._logger.warning(
+                                    f"Type mismatch for {field_name}: expected {expected_type}, got {type(value)}"
+                                )
+                                
+                            raw_args[field_name] = value
+                        else:
+                            self._logger.debug(f"Nested dict for {field_name} has no 'value' key")
+
         try:
             validated_args = self.args_schema.model_validate(raw_args)
             return validated_args.model_dump()

src/crewai/utilities/constants.py

@@ -5,14 +5,3 @@ KNOWLEDGE_DIRECTORY = "knowledge"
 MAX_LLM_RETRY = 3
 MAX_FILE_NAME_LENGTH = 255
 EMITTER_COLOR = "bold_blue"
-
-
-class _NotSpecified:
-    def __repr__(self):
-        return "NOT_SPECIFIED"
-
-
-# Sentinel value used to detect when no value has been explicitly provided.
-# Unlike `None`, which might be a valid value from the user, `NOT_SPECIFIED` allows
-# us to distinguish between "not passed at all" and "explicitly passed None" or "[]".
-NOT_SPECIFIED = _NotSpecified()

src/crewai/utilities/formatter.py

@@ -1,6 +1,6 @@
+import re
 from typing import TYPE_CHECKING, List
 
-
 if TYPE_CHECKING:
     from crewai.task import Task
     from crewai.tasks.task_output import TaskOutput
@@ -17,11 +17,6 @@ def aggregate_raw_outputs_from_task_outputs(task_outputs: List["TaskOutput"]) ->
 
 def aggregate_raw_outputs_from_tasks(tasks: List["Task"]) -> str:
     """Generate string context from the tasks."""
-
-    task_outputs = (
-        [task.output for task in tasks if task.output is not None]
-        if isinstance(tasks, list)
-        else []
-    )
+    task_outputs = [task.output for task in tasks if task.output is not None]
 
     return aggregate_raw_outputs_from_task_outputs(task_outputs)

tests/crew_test.py

@@ -2,18 +2,22 @@
 
 import hashlib
 import json
+import os
+import tempfile
 from concurrent.futures import Future
 from unittest import mock
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import MagicMock, patch
 
 import pydantic_core
 import pytest
 
 from crewai.agent import Agent
 from crewai.agents import CacheHandler
+from crewai.agents.cache import CacheHandler
+from crewai.agents.crew_agent_executor import CrewAgentExecutor
 from crewai.crew import Crew
 from crewai.crews.crew_output import CrewOutput
-from crewai.flow import Flow, start
+from crewai.flow import Flow, listen, start
 from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
 from crewai.llm import LLM
 from crewai.memory.contextual.contextual_memory import ContextualMemory
@@ -3137,30 +3141,6 @@ def test_replay_with_context():
         assert crew.tasks[1].context[0].output.raw == "context raw output"
 
 
-def test_replay_with_context_set_to_nullable():
-    agent = Agent(role="test_agent", backstory="Test Description", goal="Test Goal")
-    task1 = Task(
-        description="Context Task", expected_output="Say Task Output", agent=agent
-    )
-    task2 = Task(
-        description="Test Task", expected_output="Say Hi", agent=agent, context=[]
-    )
-    task3 = Task(
-        description="Test Task 3", expected_output="Say Hi", agent=agent, context=None
-    )
-
-    crew = Crew(agents=[agent], tasks=[task1, task2, task3], process=Process.sequential)
-    with patch("crewai.task.Task.execute_sync") as mock_execute_task:
-        mock_execute_task.return_value = TaskOutput(
-            description="Test Task Output",
-            raw="test raw output",
-            agent="test_agent",
-        )
-        crew.kickoff()
-
-    mock_execute_task.assert_called_with(agent=ANY, context="", tools=ANY)
-
-
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_replay_with_invalid_task_id():
     agent = Agent(role="test_agent", backstory="Test Description", goal="Test Goal")

tests/tools/test_structured_tool_nested_dict.py

@@ -0,0 +1,173 @@
+import pytest
+from pydantic import BaseModel, Field
+
+from crewai.tools.structured_tool import CrewStructuredTool
+
+
+class StringInputSchema(BaseModel):
+    """Schema with a string input field."""
+    query: str = Field(description="A string input parameter")
+
+
+class IntInputSchema(BaseModel):
+    """Schema with an integer input field."""
+    number: int = Field(description="An integer input parameter")
+
+
+class ComplexInputSchema(BaseModel):
+    """Schema with multiple fields of different types."""
+    text: str = Field(description="A string parameter")
+    number: int = Field(description="An integer parameter")
+    flag: bool = Field(description="A boolean parameter")
+
+
+def test_parse_args_with_string_input():
+    """Test that string inputs are parsed correctly."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test with direct string input
+    result = tool._parse_args({"query": "test string"})
+    assert result["query"] == "test string"
+    assert isinstance(result["query"], str)
+    
+    # Test with JSON string input
+    result = tool._parse_args('{"query": "json string"}')
+    assert result["query"] == "json string"
+    assert isinstance(result["query"], str)
+
+
+def test_parse_args_with_nested_dict_for_string():
+    """Test that nested dictionaries with 'value' field are handled correctly for string fields."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test with nested dict input (simulating the issue from different LLM providers)
+    nested_input = {"query": {"description": "A string input parameter", "value": "test value"}}
+    result = tool._parse_args(nested_input)
+    assert result["query"] == "test value"
+    assert isinstance(result["query"], str)
+
+
+def test_parse_args_with_nested_dict_for_int():
+    """Test that nested dictionaries with 'value' field are handled correctly for int fields."""
+    def test_func(number: int) -> str:
+        return f"Processed: {number}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="IntTool",
+        description="A tool that processes integer input"
+    )
+    
+    # Test with nested dict input for int field
+    nested_input = {"number": {"description": "An integer input parameter", "value": 42}}
+    result = tool._parse_args(nested_input)
+    assert result["number"] == 42
+    assert isinstance(result["number"], int)
+
+
+def test_parse_args_with_complex_input():
+    """Test that complex inputs with multiple fields are handled correctly."""
+    def test_func(text: str, number: int, flag: bool) -> str:
+        return f"Processed: {text}, {number}, {flag}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="ComplexTool",
+        description="A tool that processes complex input"
+    )
+    
+    # Test with mixed nested dict input
+    complex_input = {
+        "text": {"description": "A string parameter", "value": "test text"},
+        "number": 42,
+        "flag": True
+    }
+    result = tool._parse_args(complex_input)
+    assert result["text"] == "test text"
+    assert isinstance(result["text"], str)
+    assert result["number"] == 42
+    assert isinstance(result["number"], int)
+    assert result["flag"] is True
+    assert isinstance(result["flag"], bool)
+
+
+def test_invoke_with_nested_dict():
+    """Test that invoking a tool with nested dict input works correctly."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test invoking with nested dict input
+    nested_input = {"query": {"description": "A string input parameter", "value": "test value"}}
+    result = tool.invoke(nested_input)
+    assert result == "Processed: test value"
+
+
+def test_nested_dict_without_value_key():
+    """Test that nested dictionaries without 'value' field raise appropriate errors."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test with nested dict without 'value' key
+    invalid_input = {"query": {"description": "A string input parameter", "other_key": "test"}}
+    with pytest.raises(ValueError):
+        tool._parse_args(invalid_input)
+
+
+def test_empty_nested_dict():
+    """Test handling of empty nested dictionaries."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test with empty nested dict
+    empty_dict_input = {"query": {}}
+    with pytest.raises(ValueError):
+        tool._parse_args(empty_dict_input)
+
+
+def test_deeply_nested_structure():
+    """Test handling of deeply nested structures."""
+    def test_func(query: str) -> str:
+        return f"Processed: {query}"
+    
+    tool = CrewStructuredTool.from_function(
+        func=test_func,
+        name="StringTool",
+        description="A tool that processes string input"
+    )
+    
+    # Test with deeply nested structure
+    deeply_nested = {"query": {"nested": {"deeper": {"value": "deep value"}}}}
+    with pytest.raises(ValueError):
+        tool._parse_args(deeply_nested)