chore: improve CrewBase typing

This commit is contained in:
Greyson LaLonde
2025-10-21 13:58:35 -04:00
committed by GitHub
parent f3a63be215
commit 6469f224f6
3 changed files with 41 additions and 7 deletions

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from functools import wraps from functools import wraps
from typing import TYPE_CHECKING, Concatenate, ParamSpec, TypeVar from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar, overload
from crewai.project.utils import memoize from crewai.project.utils import memoize
@@ -33,6 +33,7 @@ P2 = ParamSpec("P2")
R = TypeVar("R") R = TypeVar("R")
R2 = TypeVar("R2") R2 = TypeVar("R2")
T = TypeVar("T") T = TypeVar("T")
SelfT = TypeVar("SelfT")
def before_kickoff(meth: Callable[P, R]) -> BeforeKickoffMethod[P, R]: def before_kickoff(meth: Callable[P, R]) -> BeforeKickoffMethod[P, R]:
@@ -155,9 +156,17 @@ def cache_handler(meth: Callable[P, R]) -> CacheHandlerMethod[P, R]:
return CacheHandlerMethod(memoize(meth)) return CacheHandlerMethod(memoize(meth))
@overload
def crew(
meth: Callable[Concatenate[SelfT, P], Crew],
) -> Callable[Concatenate[SelfT, P], Crew]: ...
@overload
def crew( def crew(
meth: Callable[Concatenate[CrewInstance, P], Crew], meth: Callable[Concatenate[CrewInstance, P], Crew],
) -> Callable[Concatenate[CrewInstance, P], Crew]: ) -> Callable[Concatenate[CrewInstance, P], Crew]: ...
def crew(
meth: Callable[..., Crew],
) -> Callable[..., Crew]:
"""Marks a method as the main crew execution point. """Marks a method as the main crew execution point.
Args: Args:
@@ -168,7 +177,7 @@ def crew(
""" """
@wraps(meth) @wraps(meth)
def wrapper(self: CrewInstance, *args: P.args, **kwargs: P.kwargs) -> Crew: def wrapper(self: CrewInstance, *args: Any, **kwargs: Any) -> Crew:
"""Wrapper that sets up crew before calling the decorated method. """Wrapper that sets up crew before calling the decorated method.
Args: Args:

View File

@@ -6,7 +6,15 @@ from collections.abc import Callable
import inspect import inspect
import logging import logging
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, TypeGuard, TypeVar, TypedDict, cast from typing import (
TYPE_CHECKING,
Any,
Literal,
TypeGuard,
TypeVar,
TypedDict,
cast,
)
from dotenv import load_dotenv from dotenv import load_dotenv
import yaml import yaml
@@ -320,14 +328,17 @@ def get_mcp_tools(self: CrewInstance, *tool_names: str) -> list[BaseTool]:
if not self.mcp_server_params: if not self.mcp_server_params:
return [] return []
from crewai_tools import MCPServerAdapter # type: ignore[import-untyped] from crewai_tools import MCPServerAdapter
if self._mcp_server_adapter is None: if self._mcp_server_adapter is None:
self._mcp_server_adapter = MCPServerAdapter( self._mcp_server_adapter = MCPServerAdapter(
self.mcp_server_params, connect_timeout=self.mcp_connect_timeout self.mcp_server_params, connect_timeout=self.mcp_connect_timeout
) )
return self._mcp_server_adapter.tools.filter_by_names(tool_names or None) return cast(
list[BaseTool],
self._mcp_server_adapter.tools.filter_by_names(tool_names or None),
)
def _load_config( def _load_config(
@@ -630,3 +641,17 @@ class CrewBase(metaclass=_CrewBaseType):
Note: Note:
Reference: https://stackoverflow.com/questions/11091609/setting-a-class-metaclass-using-a-decorator Reference: https://stackoverflow.com/questions/11091609/setting-a-class-metaclass-using-a-decorator
""" """
# e
if TYPE_CHECKING:
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Type stub for decorator usage.
Args:
decorated_cls: Class to transform with CrewBaseMeta metaclass.
Returns:
New class with CrewBaseMeta metaclass applied.
"""
...

View File

@@ -124,7 +124,7 @@ class CrewClass(Protocol):
get_mcp_tools: Callable[..., list[BaseTool]] get_mcp_tools: Callable[..., list[BaseTool]]
_load_config: Callable[..., dict[str, Any]] _load_config: Callable[..., dict[str, Any]]
load_configurations: Callable[..., None] load_configurations: Callable[..., None]
load_yaml: staticmethod load_yaml: Callable[..., dict[str, Any]]
map_all_agent_variables: Callable[..., None] map_all_agent_variables: Callable[..., None]
_map_agent_variables: Callable[..., None] _map_agent_variables: Callable[..., None]
map_all_task_variables: Callable[..., None] map_all_task_variables: Callable[..., None]