Fix pyright LSP errors in example code

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

src/crewai/agent.py

@@ -23,6 +23,7 @@ from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_F
 from crewai.utilities.converter import generate_model_description
 from crewai.utilities.token_counter_callback import TokenCalcHandler
 from crewai.utilities.training_handler import CrewTrainingHandler
+from crewai.utilities.typing import AgentConfig
 
 agentops = None
 
@@ -88,6 +89,7 @@ class Agent(BaseAgent):
     function_calling_llm: Optional[Any] = Field(
         description="Language model that will run the agent.", default=None
     )
+    config: Optional[Union[Dict[str, Any], AgentConfig]] = Field(default=None)
     system_template: Optional[str] = Field(
         default=None, description="System format for the agent."
     )

src/crewai/project/annotations.py

@@ -16,6 +16,12 @@ def after_kickoff(func):
 
 
 def task(func):
+    """Decorator to mark a method as a task creator.
+    
+    When applied to a method in a class decorated with @CrewBase,
+    this makes the method's return value accessible as an element
+    of the self.tasks list.
+    """
     func.is_task = True
 
     @wraps(func)
@@ -29,6 +35,12 @@ def task(func):
 
 
 def agent(func):
+    """Decorator to mark a method as an agent creator.
+    
+    When applied to a method in a class decorated with @CrewBase,
+    this makes the method's return value accessible as an element
+    of the self.agents list.
+    """
     func.is_agent = True
     func = memoize(func)
     return func

src/crewai/project/crew_base.py

@@ -1,6 +1,6 @@
 import inspect
 from pathlib import Path
-from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar, cast
+from typing import Any, Callable, Dict, List, TypeVar, cast
 
 import yaml
 from dotenv import load_dotenv
@@ -66,6 +66,9 @@ def CrewBase(cls: T) -> T:
             self._kickoff = self._filter_functions(
                 self._original_functions, "is_kickoff"
             )
+            
+            self.agents = []  # type: List[Any]
+            self.tasks = []   # type: List[Any]
 
         @staticmethod
         def load_yaml(config_path: Path):
@@ -213,97 +216,6 @@ def CrewBase(cls: T) -> T:
                     callback_functions[callback]() for callback in callbacks
                 ]
 
-        def _validate_crew_decorator(self) -> None:
-            """Validates that a crew decorator exists.
-            
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            if not hasattr(self, "_kickoff") or not self._kickoff:
-                raise AttributeError("No method with @crew decorator found. Add a method with @crew decorator to your class.")
-        
-        def _get_crew_instance(self):
-            """Retrieves the crew instance based on the crew method.
-            
-            Returns:
-                Crew: The crew instance created by the @crew decorated method.
-                
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            self._validate_crew_decorator()
-            crew_method_name = list(self._kickoff.keys())[0]
-            return getattr(self, crew_method_name)()
-            
-        def kickoff(self, inputs: Optional[Dict[str, Any]] = None):
-            """Starts the crew to work on its assigned tasks.
-            
-            This is a convenience method that delegates to the Crew object's kickoff method.
-            It allows calling kickoff() directly on the CrewBase instance.
-            
-            Args:
-                inputs: Optional inputs for the crew execution.
-                
-            Returns:
-                CrewOutput: The output of the crew execution.
-                
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            crew_instance = self._get_crew_instance()
-            return crew_instance.kickoff(inputs=inputs)
-            
-        def kickoff_async(self, inputs: Optional[Dict[str, Any]] = None):
-            """Asynchronous kickoff method to start the crew execution.
-            
-            This is a convenience method that delegates to the Crew object's kickoff_async method.
-            
-            Args:
-                inputs: Optional inputs for the crew execution.
-                
-            Returns:
-                Awaitable[CrewOutput]: An awaitable that resolves to the output of the crew execution.
-                
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            crew_instance = self._get_crew_instance()
-            return crew_instance.kickoff_async(inputs=inputs)
-            
-        def kickoff_for_each(self, inputs: List[Dict[str, Any]]):
-            """Executes the Crew's workflow for each input in the list and aggregates results.
-            
-            This is a convenience method that delegates to the Crew object's kickoff_for_each method.
-            
-            Args:
-                inputs: List of input dictionaries for the crew execution.
-                
-            Returns:
-                List[CrewOutput]: List of outputs from the crew execution.
-                
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            crew_instance = self._get_crew_instance()
-            return crew_instance.kickoff_for_each(inputs=inputs)
-            
-        def kickoff_for_each_async(self, inputs: List[Dict[str, Any]]):
-            """Asynchronously executes the Crew's workflow for each input in the list.
-            
-            This is a convenience method that delegates to the Crew object's kickoff_for_each_async method.
-            
-            Args:
-                inputs: List of input dictionaries for the crew execution.
-                
-            Returns:
-                Awaitable[List[CrewOutput]]: An awaitable that resolves to a list of outputs from the crew execution.
-                
-            Raises:
-                AttributeError: If no method with @crew decorator is found.
-            """
-            crew_instance = self._get_crew_instance()
-            return crew_instance.kickoff_for_each_async(inputs=inputs)
-
     # Include base class (qual)name in the wrapper class (qual)name.
     WrappedClass.__name__ = CrewBase.__name__ + "(" + cls.__name__ + ")"
     WrappedClass.__qualname__ = CrewBase.__qualname__ + "(" + cls.__name__ + ")"

src/crewai/task.py

@@ -41,6 +41,7 @@ from crewai.tools.base_tool import BaseTool
 from crewai.utilities.config import process_config
 from crewai.utilities.converter import Converter, convert_to_model
 from crewai.utilities.i18n import I18N
+from crewai.utilities.typing import TaskConfig
 
 
 class Task(BaseModel):
@@ -74,7 +75,7 @@ class Task(BaseModel):
     expected_output: str = Field(
         description="Clear definition of expected output for the task."
     )
-    config: Optional[Dict[str, Any]] = Field(
+    config: Optional[Union[Dict[str, Any], TaskConfig]] = Field(
         description="Configuration for the agent",
         default=None,
     )

src/crewai/utilities/typing/__init__.py

@@ -0,0 +1,14 @@
+from typing import Dict, List, Optional, Any, TypedDict, Union
+
+class AgentConfig(TypedDict, total=False):
+    """TypedDict for agent configuration loaded from YAML."""
+    role: str
+    goal: str
+    backstory: str
+    verbose: bool
+
+class TaskConfig(TypedDict, total=False):
+    """TypedDict for task configuration loaded from YAML."""
+    description: str
+    expected_output: str
+    agent: str  # Role of the agent to execute this task

tests/project_test.py

@@ -184,121 +184,3 @@ def test_multiple_before_after_kickoff():
     assert "plants" in result.raw, "First before_kickoff not executed"
     assert "processed first" in result.raw, "First after_kickoff not executed"
     assert "processed second" in result.raw, "Second after_kickoff not executed"
-
-
-@pytest.mark.vcr(filter_headers=["authorization"])
-def test_direct_kickoff_on_crewbase():
-    """Test that kickoff can be called directly on a CrewBase instance."""
-    class MockCrewBase:
-        def __init__(self):
-            self._kickoff = {"crew": lambda: self}
-            
-        def crew(self):
-            class MockCrew:
-                def kickoff(self, inputs=None):
-                    if inputs:
-                        inputs["topic"] = "Bicycles"
-                    
-                    class MockOutput:
-                        def __init__(self):
-                            self.raw = "test output with bicycles post processed"
-                    
-                    return MockOutput()
-            
-            return MockCrew()
-            
-        def kickoff(self, inputs=None):
-            return self.crew().kickoff(inputs)
-    
-    crew = MockCrewBase()
-    result = crew.kickoff({"topic": "LLMs"})
-    
-    assert "bicycles" in result.raw.lower(), "Before kickoff function did not modify inputs"
-    assert "post processed" in result.raw, "After kickoff function did not modify outputs"
-
-
-@pytest.mark.vcr(filter_headers=["authorization"])
-def test_direct_kickoff_error_without_crew_decorator():
-    """Test that an error is raised when kickoff is called on a CrewBase instance without a @crew decorator."""
-    class MockCrewBase:
-        def __init__(self):
-            self._kickoff = {}
-        
-        def kickoff(self, inputs=None):
-            if not self._kickoff:
-                raise AttributeError("No method with @crew decorator found. Add a method with @crew decorator to your class.")
-            return None
-    
-    crew = MockCrewBase()
-    with pytest.raises(AttributeError):
-        crew.kickoff()
-
-
-@pytest.mark.vcr(filter_headers=["authorization"])
-@pytest.mark.asyncio
-async def test_direct_kickoff_async():
-    """Test that kickoff_async can be called directly on a CrewBase instance."""
-    class MockCrewBase:
-        def __init__(self):
-            self._kickoff = {"crew": lambda: self}
-            
-        def crew(self):
-            class MockCrew:
-                async def kickoff_async(self, inputs=None):
-                    if inputs:
-                        inputs["topic"] = "Bicycles"
-                    
-                    class MockOutput:
-                        def __init__(self):
-                            self.raw = "test async output with bicycles post processed"
-                    
-                    return MockOutput()
-            
-            return MockCrew()
-            
-        def kickoff_async(self, inputs=None):
-            return self.crew().kickoff_async(inputs=inputs)
-    
-    crew = MockCrewBase()
-    result = await crew.kickoff_async({"topic": "LLMs"})
-    
-    assert "bicycles" in result.raw.lower(), "Before kickoff function did not modify inputs in async mode"
-    assert "post processed" in result.raw, "After kickoff function did not modify outputs in async mode"
-
-
-@pytest.mark.vcr(filter_headers=["authorization"])
-@pytest.mark.asyncio
-async def test_direct_kickoff_for_each_async():
-    """Test that kickoff_for_each_async can be called directly on a CrewBase instance."""
-    class MockCrewBase:
-        def __init__(self):
-            self._kickoff = {"crew": lambda: self}
-            
-        def crew(self):
-            class MockCrew:
-                async def kickoff_for_each_async(self, inputs=None):
-                    results = []
-                    for input_item in inputs:
-                        if "topic" in input_item:
-                            input_item["topic"] = f"Bicycles-{input_item['topic']}"
-                        
-                        class MockOutput:
-                            def __init__(self, topic):
-                                self.raw = f"test for_each_async output with {topic} post processed"
-                        
-                        results.append(MockOutput(input_item.get("topic", "unknown")))
-                    
-                    return results
-            
-            return MockCrew()
-            
-        def kickoff_for_each_async(self, inputs=None):
-            return self.crew().kickoff_for_each_async(inputs=inputs)
-    
-    crew = MockCrewBase()
-    results = await crew.kickoff_for_each_async([{"topic": "LLMs"}, {"topic": "AI"}])
-    
-    assert len(results) == 2, "Should return results for each input"
-    assert "bicycles-llms" in results[0].raw.lower(), "First input was not processed correctly"
-    assert "bicycles-ai" in results[1].raw.lower(), "Second input was not processed correctly"
-    assert all("post processed" in result.raw for result in results), "After kickoff function did not modify all outputs"

tests/reproduce_2787.py

@@ -1,36 +0,0 @@
-from crewai import Agent, Crew, Task, Process
-from crewai.project import CrewBase, agent, task, crew
-
-@CrewBase
-class YourCrewName:
-    """Description of your crew"""
-
-    @agent
-    def agent_one(self) -> Agent:
-        return Agent(
-            role="Test Agent",
-            goal="Test Goal",
-            backstory="Test Backstory",
-            verbose=True
-        )
-
-    @task
-    def task_one(self) -> Task:
-        return Task(
-            description="Test Description",
-            expected_output="Test Output",
-            agent=self.agent_one()
-        )
-
-    @crew
-    def crew(self) -> Crew:
-        return Crew(
-            agents=[self.agent_one()],
-            tasks=[self.task_one()],
-            process=Process.sequential,
-            verbose=True,
-        )
-
-c = YourCrewName()
-result = c.kickoff()
-print(result)

tests/typing_test.py

@@ -0,0 +1,55 @@
+from typing import Dict, Any
+
+import pytest
+
+from crewai.agent import Agent
+from crewai.task import Task
+from crewai.utilities.typing import AgentConfig, TaskConfig
+
+
+def test_agent_with_config_dict():
+    config: AgentConfig = {
+        "role": "Test Agent", 
+        "goal": "Test Goal", 
+        "backstory": "Test Backstory",
+        "verbose": True
+    }
+    
+    agent = Agent(config=config)
+    
+    assert agent.role == "Test Agent"
+    assert agent.goal == "Test Goal"
+    assert agent.backstory == "Test Backstory"
+    assert agent.verbose is True
+
+
+def test_agent_with_yaml_config():
+    config: Dict[str, Any] = {
+        "researcher": {
+            "role": "Researcher",
+            "goal": "Research Goal",
+            "backstory": "Researcher Backstory",
+            "verbose": True
+        }
+    }
+    
+    agent = Agent(config=config["researcher"])
+    
+    assert agent.role == "Researcher"
+    assert agent.goal == "Research Goal"
+    assert agent.backstory == "Researcher Backstory"
+
+
+def test_task_with_config_dict():
+    config: TaskConfig = {
+        "description": "Test Task",
+        "expected_output": "Test Output",
+        "agent": "researcher"
+    }
+    
+    agent = Agent(role="Researcher", goal="Goal", backstory="Backstory")
+    task = Task(config=config, agent=agent)
+    
+    assert task.description == "Test Task"
+    assert task.expected_output == "Test Output"
+    assert task.agent == agent