Add explicit type annotations for agents_config and tasks_config

Fixes #3801

This commit addresses the issue where agents_config and tasks_config
in CrewBase-decorated classes lacked explicit type annotations, making
the code un-Pythonic and reducing IDE/type checker support.

Changes:
- Added AgentsConfigDict and TasksConfigDict type aliases
- Updated load_configurations to use cast() for proper typing
- Updated CrewInstance Protocol to use typed config dictionaries
- Exported type aliases from crewai.project for user convenience
- Updated TypedDicts to support both raw YAML and processed values
- Added comprehensive tests for config loading and type annotations

The type aliases allow users to import and use proper types:
  from crewai.project import AgentConfig, AgentsConfigDict

This provides better IDE autocomplete and static type checking when
accessing self.agents_config and self.tasks_config in CrewBase classes.

Co-Authored-By: João <joao@crewai.com>
This commit is contained in:
Devin AI
2025-10-27 10:49:01 +00:00
parent 494ed7e671
commit 1784636e93
6 changed files with 4273 additions and 4027 deletions

View File

@@ -13,11 +13,21 @@ from crewai.project.annotations import (
task,
tool,
)
from crewai.project.crew_base import CrewBase
from crewai.project.crew_base import (
AgentConfig,
AgentsConfigDict,
CrewBase,
TaskConfig,
TasksConfigDict,
)
__all__ = [
"AgentConfig",
"AgentsConfigDict",
"CrewBase",
"TaskConfig",
"TasksConfigDict",
"after_kickoff",
"agent",
"before_kickoff",

View File

@@ -52,11 +52,11 @@ class AgentConfig(TypedDict, total=False):
allow_delegation: bool
max_iter: int
max_tokens: int
callbacks: list[str]
callbacks: list[str] | list[Any]
# LLM configuration
llm: str
function_calling_llm: str
# LLM configuration (can be string references or resolved instances)
llm: str | Any
function_calling_llm: str | Any
use_system_prompt: bool
# Template configuration
@@ -66,7 +66,7 @@ class AgentConfig(TypedDict, total=False):
# Tools and handlers (can be string references or instances)
tools: list[str] | list[BaseTool]
step_callback: str
step_callback: str | Any
cache_handler: str | CacheHandler
# Code execution
@@ -111,18 +111,18 @@ class TaskConfig(TypedDict, total=False):
description: str
expected_output: str
# Agent and context
agent: str
context: list[str]
# Agent and context (can be string references or resolved instances)
agent: str | Any
context: list[str] | list[Any]
# Tools and callbacks (can be string references or instances)
tools: list[str] | list[BaseTool]
callback: str
callbacks: list[str]
callback: str | Any
callbacks: list[str] | list[Any]
# Output configuration
output_json: str
output_pydantic: str
# Output configuration (can be string references or resolved class wrappers)
output_json: str | Any
output_pydantic: str | Any
output_file: str
create_directory: bool
@@ -139,6 +139,10 @@ class TaskConfig(TypedDict, total=False):
allow_crewai_trigger_context: bool
AgentsConfigDict = dict[str, AgentConfig]
TasksConfigDict = dict[str, TaskConfig]
load_dotenv()
CallableT = TypeVar("CallableT", bound=Callable[..., Any])
@@ -378,8 +382,14 @@ def load_configurations(self: CrewInstance) -> None:
Args:
self: Crew instance with configuration paths.
"""
self.agents_config = self._load_config(self.original_agents_config_path, "agent")
self.tasks_config = self._load_config(self.original_tasks_config_path, "task")
self.agents_config = cast(
AgentsConfigDict,
self._load_config(self.original_agents_config_path, "agent"),
)
self.tasks_config = cast(
TasksConfigDict,
self._load_config(self.original_tasks_config_path, "task"),
)
def load_yaml(config_path: Path) -> dict[str, Any]:

View File

@@ -22,6 +22,12 @@ from typing_extensions import Self
if TYPE_CHECKING:
from crewai import Agent, Crew, Task
from crewai.crews.crew_output import CrewOutput
from crewai.project.crew_base import (
AgentConfig,
AgentsConfigDict,
TaskConfig,
TasksConfigDict,
)
from crewai.tools import BaseTool
@@ -75,8 +81,8 @@ class CrewInstance(Protocol):
base_directory: Path
original_agents_config_path: str
original_tasks_config_path: str
agents_config: dict[str, Any]
tasks_config: dict[str, Any]
agents_config: AgentsConfigDict
tasks_config: TasksConfigDict
mcp_server_params: Any
mcp_connect_timeout: int
@@ -90,7 +96,7 @@ class CrewInstance(Protocol):
def _map_agent_variables(
self,
agent_name: str,
agent_info: dict[str, Any],
agent_info: AgentConfig,
llms: dict[str, Callable[..., Any]],
tool_functions: dict[str, Callable[..., Any]],
cache_handler_functions: dict[str, Callable[..., Any]],
@@ -99,7 +105,7 @@ class CrewInstance(Protocol):
def _map_task_variables(
self,
task_name: str,
task_info: dict[str, Any],
task_info: TaskConfig,
agents: dict[str, Callable[..., Any]],
tasks: dict[str, Callable[..., Any]],
output_json_functions: dict[str, Callable[..., Any]],

View File

View File

@@ -0,0 +1,193 @@
"""Tests for CrewBase configuration type annotations."""
from pathlib import Path
import pytest
from crewai.project import AgentConfig, AgentsConfigDict, CrewBase, TaskConfig, TasksConfigDict, agent, task
def test_agents_config_loads_as_dict(tmp_path: Path) -> None:
"""Test that agents_config loads as a properly typed dictionary."""
agents_yaml = tmp_path / "agents.yaml"
agents_yaml.write_text(
"""
researcher:
role: "Research Analyst"
goal: "Find accurate information"
backstory: "Expert researcher with years of experience"
"""
)
tasks_yaml = tmp_path / "tasks.yaml"
tasks_yaml.write_text(
"""
research_task:
description: "Research the topic"
expected_output: "A comprehensive report"
"""
)
@CrewBase
class TestCrew:
agents_config = str(agents_yaml)
tasks_config = str(tasks_yaml)
@agent
def researcher(self):
from crewai import Agent
return Agent(config=self.agents_config["researcher"])
@task
def research_task(self):
from crewai import Task
return Task(config=self.tasks_config["research_task"])
crew_instance = TestCrew()
assert isinstance(crew_instance.agents_config, dict)
assert "researcher" in crew_instance.agents_config
assert crew_instance.agents_config["researcher"]["role"] == "Research Analyst"
assert crew_instance.agents_config["researcher"]["goal"] == "Find accurate information"
assert crew_instance.agents_config["researcher"]["backstory"] == "Expert researcher with years of experience"
def test_tasks_config_loads_as_dict(tmp_path: Path) -> None:
"""Test that tasks_config loads as a properly typed dictionary."""
agents_yaml = tmp_path / "agents.yaml"
agents_yaml.write_text(
"""
writer:
role: "Content Writer"
goal: "Write engaging content"
backstory: "Experienced content writer"
"""
)
tasks_yaml = tmp_path / "tasks.yaml"
tasks_yaml.write_text(
"""
writing_task:
description: "Write an article"
expected_output: "A well-written article"
agent: "writer"
"""
)
@CrewBase
class TestCrew:
agents_config = str(agents_yaml)
tasks_config = str(tasks_yaml)
@agent
def writer(self):
from crewai import Agent
return Agent(config=self.agents_config["writer"])
@task
def writing_task(self):
from crewai import Task
return Task(config=self.tasks_config["writing_task"])
crew_instance = TestCrew()
assert isinstance(crew_instance.tasks_config, dict)
assert "writing_task" in crew_instance.tasks_config
assert crew_instance.tasks_config["writing_task"]["description"] == "Write an article"
assert crew_instance.tasks_config["writing_task"]["expected_output"] == "A well-written article"
from crewai import Agent
assert isinstance(crew_instance.tasks_config["writing_task"]["agent"], Agent)
assert crew_instance.tasks_config["writing_task"]["agent"].role == "Content Writer"
def test_empty_config_files_load_as_empty_dicts(tmp_path: Path) -> None:
"""Test that empty config files load as empty dictionaries."""
agents_yaml = tmp_path / "agents.yaml"
agents_yaml.write_text("")
tasks_yaml = tmp_path / "tasks.yaml"
tasks_yaml.write_text("")
@CrewBase
class TestCrew:
agents_config = str(agents_yaml)
tasks_config = str(tasks_yaml)
crew_instance = TestCrew()
assert isinstance(crew_instance.agents_config, dict)
assert isinstance(crew_instance.tasks_config, dict)
assert len(crew_instance.agents_config) == 0
assert len(crew_instance.tasks_config) == 0
def test_missing_config_files_load_as_empty_dicts(tmp_path: Path) -> None:
"""Test that missing config files load as empty dictionaries with warning."""
nonexistent_agents = tmp_path / "nonexistent_agents.yaml"
nonexistent_tasks = tmp_path / "nonexistent_tasks.yaml"
@CrewBase
class TestCrew:
agents_config = str(nonexistent_agents)
tasks_config = str(nonexistent_tasks)
crew_instance = TestCrew()
assert isinstance(crew_instance.agents_config, dict)
assert isinstance(crew_instance.tasks_config, dict)
assert len(crew_instance.agents_config) == 0
assert len(crew_instance.tasks_config) == 0
def test_config_types_are_exported() -> None:
"""Test that AgentConfig, TaskConfig, and type aliases are properly exported."""
from crewai.project import AgentConfig, AgentsConfigDict, TaskConfig, TasksConfigDict
assert AgentConfig is not None
assert TaskConfig is not None
assert AgentsConfigDict is not None
assert TasksConfigDict is not None
def test_agents_config_type_annotation_exists(tmp_path: Path) -> None:
"""Test that agents_config has proper type annotation at runtime."""
agents_yaml = tmp_path / "agents.yaml"
agents_yaml.write_text(
"""
analyst:
role: "Data Analyst"
goal: "Analyze data"
"""
)
tasks_yaml = tmp_path / "tasks.yaml"
tasks_yaml.write_text(
"""
analysis:
description: "Analyze the data"
expected_output: "Analysis report"
"""
)
@CrewBase
class TestCrew:
agents_config = str(agents_yaml)
tasks_config = str(tasks_yaml)
@agent
def analyst(self):
from crewai import Agent
return Agent(config=self.agents_config["analyst"])
@task
def analysis(self):
from crewai import Task
return Task(config=self.tasks_config["analysis"])
crew_instance = TestCrew()
assert hasattr(crew_instance, "agents_config")
assert hasattr(crew_instance, "tasks_config")
assert isinstance(crew_instance.agents_config, dict)
assert isinstance(crew_instance.tasks_config, dict)

8041
uv.lock generated

File diff suppressed because it is too large Load Diff