from __future__ import annotations from abc import ABC, abstractmethod from collections.abc import Sequence from copy import copy as shallow_copy from hashlib import md5 from pathlib import Path import re from typing import TYPE_CHECKING, Annotated, Any, Final, Literal import uuid from pydantic import ( UUID4, BaseModel, BeforeValidator, Field, PrivateAttr, SerializeAsAny, field_validator, model_validator, ) from pydantic.functional_serializers import PlainSerializer from pydantic_core import PydanticCustomError from typing_extensions import Self from crewai.agent.internal.meta import AgentMeta from crewai.agents.agent_builder.base_agent_executor import BaseAgentExecutor from crewai.agents.agent_builder.utilities.base_token_process import TokenProcess from crewai.agents.cache.cache_handler import CacheHandler from crewai.agents.tools_handler import ToolsHandler from crewai.events.base_events import set_emission_counter from crewai.events.event_bus import crewai_event_bus from crewai.events.event_context import restore_event_scope, set_last_event_id from crewai.knowledge.knowledge import Knowledge, _resolve_knowledge_sources from crewai.knowledge.knowledge_config import KnowledgeConfig from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource from crewai.knowledge.storage.base_knowledge_storage import BaseKnowledgeStorage from crewai.llms.base_llm import BaseLLM from crewai.mcp.config import MCPServerConfig from crewai.memory.memory_scope import MemoryScope, MemorySlice, _ensure_memory_kind from crewai.memory.unified_memory import Memory from crewai.rag.embeddings.types import EmbedderConfig from crewai.security.security_config import SecurityConfig from crewai.skills.models import Skill from crewai.state.checkpoint_config import CheckpointConfig, _coerce_checkpoint from crewai.tools.base_tool import BaseTool, Tool from crewai.types.callback import SerializableCallable from crewai.utilities.config import process_config from crewai.utilities.i18n import I18N, get_i18n from crewai.utilities.logger import Logger from crewai.utilities.rpm_controller import RPMController from crewai.utilities.string_utils import interpolate_only if TYPE_CHECKING: from crewai.context import ExecutionContext from crewai.crew import Crew from crewai.state.runtime import RuntimeState def _validate_crew_ref(value: Any) -> Any: return value def _serialize_crew_ref(value: Any) -> str | None: if value is None: return None return str(value.id) if hasattr(value, "id") else str(value) _LLM_TYPE_REGISTRY: dict[str, str] = { "base": "crewai.llms.base_llm.BaseLLM", "litellm": "crewai.llm.LLM", "openai": "crewai.llms.providers.openai.completion.OpenAICompletion", "anthropic": "crewai.llms.providers.anthropic.completion.AnthropicCompletion", "azure": "crewai.llms.providers.azure.completion.AzureCompletion", "bedrock": "crewai.llms.providers.bedrock.completion.BedrockCompletion", "gemini": "crewai.llms.providers.gemini.completion.GeminiCompletion", } def _validate_llm_ref(value: Any) -> Any: if isinstance(value, dict): import importlib import inspect llm_type = value.get("llm_type") if not llm_type: model = ( value.get("model") or value.get("model_name") or value.get("deployment_name") ) if not model: raise ValueError( "LLM config objects must include 'model', 'model_name', " "or 'deployment_name', or a serialized 'llm_type'. " f"Got keys: {list(value)}" ) from crewai.llm import LLM llm_kwargs = {**value, "model": model} llm_kwargs.pop("model_name", None) llm_kwargs.pop("deployment_name", None) return LLM(**llm_kwargs) if llm_type not in _LLM_TYPE_REGISTRY: raise ValueError( f"Unknown llm_type: {llm_type!r}. " f"Expected one of {list(_LLM_TYPE_REGISTRY)}" ) dotted = _LLM_TYPE_REGISTRY[llm_type] mod_path, cls_name = dotted.rsplit(".", 1) cls = getattr(importlib.import_module(mod_path), cls_name) if inspect.isabstract(cls): from crewai.llm import LLM return LLM( **{k: v for k, v in value.items() if v is not None and k != "llm_type"} ) return cls(**value) return value def _resolve_agent(value: Any, info: Any) -> Any: if isinstance(value, BaseAgent) or value is None or not isinstance(value, dict): return value from crewai.agent.core import Agent return Agent.model_validate(value, context=getattr(info, "context", None)) _EXECUTOR_TYPE_REGISTRY: dict[str, str] = { "base": "crewai.agents.agent_builder.base_agent_executor.BaseAgentExecutor", "crew": "crewai.agents.crew_agent_executor.CrewAgentExecutor", "experimental": "crewai.experimental.agent_executor.AgentExecutor", } def _validate_executor_ref(value: Any) -> Any: if isinstance(value, dict): import importlib executor_type = value.get("executor_type") if not executor_type or executor_type not in _EXECUTOR_TYPE_REGISTRY: raise ValueError( f"Unknown or missing executor_type: {executor_type!r}. " f"Expected one of {list(_EXECUTOR_TYPE_REGISTRY)}" ) dotted = _EXECUTOR_TYPE_REGISTRY[executor_type] mod_path, cls_name = dotted.rsplit(".", 1) cls = getattr(importlib.import_module(mod_path), cls_name) return cls.model_validate(value) return value def _serialize_executor_ref(value: Any) -> dict[str, Any] | None: if value is None: return None result: dict[str, Any] = value.model_dump(mode="json") return result def _serialize_llm_ref(value: Any) -> dict[str, Any] | None: if value is None: return None if isinstance(value, str): return {"model": value} result: dict[str, Any] = value.model_dump() return result _SLUG_RE: Final[re.Pattern[str]] = re.compile( r"^(?:crewai-amp:)?[a-zA-Z0-9][a-zA-Z0-9_-]*(?:#[\w-]+)?$" ) PlatformApp = Literal[ "asana", "box", "clickup", "github", "gmail", "google_calendar", "google_sheets", "hubspot", "jira", "linear", "notion", "salesforce", "shopify", "slack", "stripe", "zendesk", ] PlatformAppOrAction = PlatformApp | str class BaseAgent(BaseModel, ABC, metaclass=AgentMeta): """Abstract Base Class for all third party agents compatible with CrewAI. Attributes: id (UUID4): Unique identifier for the agent. role (str): Role of the agent. goal (str): Objective of the agent. backstory (str): Backstory of the agent. cache (bool): Whether the agent participates in tool-result caching when a cache is enabled. The default (True) only permits participation — caching activates when the crew sets cache=True or the agent explicitly opts in with cache=True or a cache_handler; cache=False excludes the agent entirely. config (dict[str, Any] | None): Configuration for the agent. verbose (bool): Verbose mode for the Agent Execution. max_rpm (int | None): Maximum number of requests per minute for the agent execution. allow_delegation (bool): Allow delegation of tasks to agents. tools (list[Any] | None): Tools at the agent's disposal. max_iter (int): Maximum iterations for an agent to execute a task. agent_executor: An instance of the CrewAgentExecutor class. i18n (I18N): Internationalization settings. llm (Any): Language model that will run the agent. crew (Any): Crew to which the agent belongs. cache_handler ([CacheHandler]): An instance of the CacheHandler class. tools_handler ([ToolsHandler]): An instance of the ToolsHandler class. max_tokens: Maximum number of tokens for the agent to generate in a response. knowledge_sources: Knowledge sources for the agent. knowledge_storage: Custom knowledge storage for the agent. security_config: Security configuration for the agent, including fingerprinting. apps: List of enterprise applications that the agent can access through CrewAI AMP Tools. Methods: execute_task(task: Any, context: str | None = None, tools: list[BaseTool] | None = None) -> str: Abstract method to execute a task. create_agent_executor(tools=None) -> None: Abstract method to create an agent executor. get_delegation_tools(agents: list["BaseAgent"]): Abstract method to set the agents task tools for handling delegation and question asking to other agents in crew. get_platform_tools(apps: list[PlatformAppOrAction]): Abstract method to get platform tools for the specified list of applications and/or application/action combinations. get_output_converter(llm, model, instructions): Abstract method to get the converter class for the agent to create json/pydantic outputs. interpolate_inputs(inputs: dict[str, Any]) -> None: Interpolate inputs into the agent description and backstory. set_cache_handler(cache_handler: CacheHandler) -> None: Set the cache handler for the agent. copy() -> "BaseAgent": Create a copy of the agent. set_rpm_controller(rpm_controller: RPMController) -> None: Set the rpm controller for the agent. set_private_attrs() -> "BaseAgent": Set private attributes. """ entity_type: Literal["agent"] = "agent" __hash__ = object.__hash__ _logger: Logger = PrivateAttr(default_factory=lambda: Logger(verbose=False)) _rpm_controller: RPMController | None = PrivateAttr(default=None) _request_within_rpm_limit: SerializableCallable | None = PrivateAttr(default=None) _constructor_cache_opt_in: bool = PrivateAttr(default=False) _original_role: str | None = PrivateAttr(default=None) _original_goal: str | None = PrivateAttr(default=None) _original_backstory: str | None = PrivateAttr(default=None) _token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess) _kickoff_event_id: str | None = PrivateAttr(default=None) id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True) role: str = Field(description="Role of the agent") goal: str = Field(description="Objective of the agent") backstory: str = Field(description="Backstory of the agent") config: dict[str, Any] | None = Field( description="Configuration for the agent", default=None, exclude=True ) cache: bool = Field( default=True, description=( "Whether the agent participates in tool-result caching when a " "cache is enabled. Caching itself is opt-in: it activates only " "when the crew sets cache=True or the agent explicitly opts in " "(cache=True or a cache_handler at construction). Set False to " "exclude this agent even when the crew enables caching." ), ) verbose: bool = Field( default=False, description="Verbose mode for the Agent Execution" ) max_rpm: int | None = Field( default=None, description="Maximum number of requests per minute for the agent execution to be respected.", ) allow_delegation: bool = Field( default=False, description="Enable agent to delegate and ask questions among each other.", ) tools: list[BaseTool] | None = Field( default_factory=list, description="Tools at agents' disposal" ) max_iter: int = Field( default=25, description="Maximum iterations for an agent to execute a task" ) agent_executor: Annotated[ SerializeAsAny[BaseAgentExecutor] | None, BeforeValidator(_validate_executor_ref), PlainSerializer( _serialize_executor_ref, return_type=dict | None, when_used="json" ), ] = Field(default=None, description="An instance of the CrewAgentExecutor class.") i18n: I18N = Field( default_factory=get_i18n, description="Internationalization settings.", deprecated=( "Agent.i18n is deprecated and will be removed in a future release. " "Use crewai.utilities.i18n.get_i18n() or Crew(prompt_file=...) instead." ), ) llm: Annotated[ str | BaseLLM | None, BeforeValidator(_validate_llm_ref), PlainSerializer(_serialize_llm_ref, return_type=dict | None, when_used="json"), ] = Field(default=None, description="Language model that will run the agent.") crew: Annotated[ Crew | str | None, BeforeValidator(_validate_crew_ref), PlainSerializer( _serialize_crew_ref, return_type=str | None, when_used="always" ), ] = Field(default=None, description="Crew to which the agent belongs.") cache_handler: CacheHandler | None = Field( default=None, description="An instance of the CacheHandler class." ) tools_handler: ToolsHandler = Field( default_factory=ToolsHandler, description="An instance of the ToolsHandler class.", ) tools_results: list[dict[str, Any]] = Field( default_factory=list, description="Results of the tools used by the agent." ) max_tokens: int | None = Field( default=None, description="Maximum number of tokens for the agent's execution." ) knowledge: Knowledge | None = Field( default=None, description="Knowledge for the agent." ) knowledge_sources: Annotated[ list[BaseKnowledgeSource] | None, BeforeValidator(_resolve_knowledge_sources), ] = Field( default=None, description="Knowledge sources for the agent.", ) knowledge_storage: BaseKnowledgeStorage | None = Field( default=None, description="Custom knowledge storage for the agent.", ) security_config: SecurityConfig = Field( default_factory=SecurityConfig, description="Security configuration for the agent, including fingerprinting.", ) checkpoint: Annotated[ CheckpointConfig | bool | None, BeforeValidator(_coerce_checkpoint), ] = Field( default=None, description="Automatic checkpointing configuration. " "True for defaults, False to opt out, None to inherit.", ) callbacks: list[SerializableCallable] = Field( default_factory=list, description="Callbacks to be used for the agent" ) adapted_agent: bool = Field( default=False, description="Whether the agent is adapted" ) knowledge_config: KnowledgeConfig | None = Field( default=None, description="Knowledge configuration for the agent such as limits and threshold", ) apps: list[PlatformAppOrAction] | None = Field( default=None, description="List of applications or application/action combinations that the agent can access through CrewAI Platform. Can contain app names (e.g., 'gmail') or specific actions (e.g., 'gmail/send_email')", ) mcps: list[str | MCPServerConfig] | None = Field( default=None, description="List of MCP server references. Supports 'https://server.com/path' for external servers and bare slugs like 'notion' for connected MCP integrations. Use '#tool_name' suffix for specific tools.", ) memory: Annotated[ bool | Annotated[ Memory | MemoryScope | MemorySlice, Field(discriminator="memory_kind") ] | None, BeforeValidator(_ensure_memory_kind), ] = Field( default=None, description=( "Enable agent memory. Pass True for default Memory(), " "or a Memory/MemoryScope/MemorySlice instance for custom configuration. " "If not set, falls back to crew memory." ), ) skills: list[Path | Skill | str] | None = Field( default=None, description="Agent Skills. Accepts paths for discovery, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs.", min_length=1, ) execution_context: ExecutionContext | None = Field(default=None) checkpoint_kickoff_event_id: str | None = Field(default=None) @classmethod def from_checkpoint(cls, config: CheckpointConfig) -> Self: """Restore an Agent from a checkpoint, ready to resume via kickoff(). Args: config: Checkpoint configuration with ``restore_from`` set to the path of the checkpoint to load. Returns: An Agent instance. Call kickoff() to resume execution. """ from crewai.context import apply_execution_context from crewai.state.runtime import RuntimeState state = RuntimeState.from_checkpoint(config, context={"from_checkpoint": True}) crewai_event_bus.set_runtime_state(state) for entity in state.root: if isinstance(entity, cls): if entity.execution_context is not None: apply_execution_context(entity.execution_context) entity._restore_runtime(state) return entity raise ValueError( f"No {cls.__name__} found in checkpoint: {config.restore_from}" ) @classmethod def fork(cls, config: CheckpointConfig, branch: str | None = None) -> Self: """Fork an Agent from a checkpoint, creating a new execution branch. Args: config: Checkpoint configuration with ``restore_from`` set. branch: Branch label for the fork. Auto-generated if not provided. Returns: An Agent instance on the new branch. Call kickoff() to run. """ agent = cls.from_checkpoint(config) state = crewai_event_bus._runtime_state if state is None: raise RuntimeError("Cannot fork: no runtime state on the event bus.") state.fork(branch) return agent def _restore_runtime(self, state: RuntimeState) -> None: """Re-create runtime objects after restoring from a checkpoint. Args: state: The RuntimeState containing the event record. """ if self.agent_executor is not None: self.agent_executor.agent = self self.agent_executor._resuming = True if self.checkpoint_kickoff_event_id is not None: self._kickoff_event_id = self.checkpoint_kickoff_event_id self._rebind_memory_view() self._restore_event_scope(state) def _rebind_memory_view(self) -> None: """Reattach a fresh ``Memory`` to a restored ``MemoryScope``/``MemorySlice``. Checkpoint JSON omits the live ``Memory`` dependency, so scoped memory views raise ``RuntimeError`` on first use after restore. """ if ( isinstance(self.memory, MemoryScope | MemorySlice) and self.memory._memory is None ): self.memory.bind(Memory()) def _restore_event_scope(self, state: RuntimeState) -> None: """Rebuild the event scope stack from the checkpoint's event record. Args: state: The RuntimeState containing the event record. """ stack: list[tuple[str, str]] = [] kickoff_id = self._kickoff_event_id if kickoff_id: stack.append((kickoff_id, "lite_agent_execution_started")) restore_event_scope(tuple(stack)) last_event_id: str | None = None max_seq = 0 for node in state.event_record.nodes.values(): seq = node.event.emission_sequence or 0 if seq > max_seq: max_seq = seq last_event_id = node.event.event_id if last_event_id is not None: set_last_event_id(last_event_id) if max_seq > 0: set_emission_counter(max_seq) @model_validator(mode="before") @classmethod def process_model_config(cls, values: Any) -> dict[str, Any]: return process_config(values, cls) @field_validator("tools") @classmethod def validate_tools(cls, tools: list[Any]) -> list[BaseTool]: """Validate and process the tools provided to the agent. This method ensures that each tool is either an instance of BaseTool or an object with 'name', 'func', and 'description' attributes. If the tool meets these criteria, it is processed and added to the list of tools. Otherwise, a ValueError is raised. """ if not tools: return [] processed_tools = [] required_attrs = ["name", "func", "description"] for tool in tools: if isinstance(tool, BaseTool): processed_tools.append(tool) elif all(hasattr(tool, attr) for attr in required_attrs): processed_tools.append(Tool.from_langchain(tool)) else: raise ValueError( f"Invalid tool type: {type(tool)}. " "Tool must be an instance of BaseTool or " "an object with 'name', 'func', and 'description' attributes." ) return processed_tools @field_validator("apps") @classmethod def validate_apps( cls, apps: list[PlatformAppOrAction] | None ) -> list[PlatformAppOrAction] | None: if not apps: return apps validated_apps = [] for app in apps: if app.count("/") > 1: raise ValueError( f"Invalid app format '{app}'. Apps can only have one '/' for app/action format (e.g., 'gmail/send_email')" ) validated_apps.append(app) return list(set(validated_apps)) @field_validator("mcps") @classmethod def validate_mcps( cls, mcps: list[str | MCPServerConfig] | None ) -> list[str | MCPServerConfig] | None: """Validate MCP server references and configurations. Supports both string references (for backwards compatibility) and structured configuration objects (MCPServerStdio, MCPServerHTTP, MCPServerSSE). """ if not mcps: return mcps validated_mcps: list[str | MCPServerConfig] = [] for mcp in mcps: if isinstance(mcp, str): if mcp.startswith("https://"): validated_mcps.append(mcp) elif _SLUG_RE.match(mcp): validated_mcps.append(mcp) else: raise ValueError( f"Invalid MCP reference: {mcp!r}. " "String references must be an 'https://' URL or a valid " "slug (e.g. 'notion', 'notion#search', 'crewai-amp:notion')." ) elif isinstance(mcp, (MCPServerConfig)): validated_mcps.append(mcp) else: raise ValueError( f"Invalid MCP configuration: {type(mcp)}. " "Must be a string reference or MCPServerConfig instance." ) return validated_mcps @model_validator(mode="after") def validate_and_set_attributes(self) -> Self: for field in ["role", "goal", "backstory"]: if getattr(self, field) is None: raise ValueError( f"{field} must be provided either directly or through config" ) self._logger = Logger(verbose=self.verbose) if self.max_rpm and not self._rpm_controller: self._rpm_controller = RPMController( max_rpm=self.max_rpm, logger=self._logger ) if not self._token_process: self._token_process = TokenProcess() if self.security_config is None: self.security_config = SecurityConfig() return self @field_validator("id", mode="before") @classmethod def _deny_user_set_id(cls, v: UUID4 | None, info: Any) -> UUID4 | None: if v and not (info.context or {}).get("from_checkpoint"): raise PydanticCustomError( "may_not_set_field", "This field is not to be set by the user.", {} ) return v @model_validator(mode="after") def set_private_attrs(self) -> Self: """Set private attributes.""" self._logger = Logger(verbose=self.verbose) if self.max_rpm and not self._rpm_controller: self._rpm_controller = RPMController( max_rpm=self.max_rpm, logger=self._logger ) if not self._token_process: self._token_process = TokenProcess() return self @model_validator(mode="after") def resolve_memory(self) -> Self: """Resolve memory field: True creates a default Memory(), instance is used as-is.""" if self.memory is True: from crewai.memory.unified_memory import Memory memory_kwargs: dict[str, Any] = {} if self.llm is not None: memory_kwargs["llm"] = self.llm self.memory = Memory(**memory_kwargs) elif self.memory is False: self.memory = None return self @property def key(self) -> str: source = [ self._original_role or self.role, self._original_goal or self.goal, self._original_backstory or self.backstory, ] return md5("|".join(source).encode(), usedforsecurity=False).hexdigest() @abstractmethod def execute_task( self, task: Any, context: str | None = None, tools: list[BaseTool] | None = None, ) -> str: pass @abstractmethod async def aexecute_task( self, task: Any, context: str | None = None, tools: list[BaseTool] | None = None, ) -> str: """Execute a task asynchronously.""" @abstractmethod def create_agent_executor(self, tools: list[BaseTool] | None = None) -> None: pass @abstractmethod def get_delegation_tools(self, agents: Sequence[BaseAgent]) -> list[BaseTool]: """Set the task tools that init BaseAgenTools class.""" @abstractmethod def get_platform_tools(self, apps: list[PlatformAppOrAction]) -> list[BaseTool]: """Get platform tools for the specified list of applications and/or application/action combinations.""" @abstractmethod def get_mcp_tools(self, mcps: list[str | MCPServerConfig]) -> list[BaseTool]: """Get MCP tools for the specified list of MCP server references.""" def copy(self) -> Self: # type: ignore # Signature of "copy" incompatible with supertype "BaseModel" """Create a deep copy of the Agent.""" exclude = { "id", "_logger", "_rpm_controller", "_request_within_rpm_limit", "_token_process", "agent_executor", "tools", "tools_handler", "cache_handler", "llm", "knowledge_sources", "knowledge_storage", "knowledge", "apps", "mcps", "actions", } existing_llm = shallow_copy(self.llm) copied_knowledge = shallow_copy(self.knowledge) copied_knowledge_storage = shallow_copy(self.knowledge_storage) existing_knowledge_sources = None if self.knowledge_sources: shared_storage = ( self.knowledge_sources[0].storage if self.knowledge_sources else None ) existing_knowledge_sources = [] for source in self.knowledge_sources: copied_source = ( source.model_copy() if hasattr(source, "model_copy") else shallow_copy(source) ) copied_source.storage = shared_storage existing_knowledge_sources.append(copied_source) copied_data = self.model_dump(exclude=exclude) copied_data = {k: v for k, v in copied_data.items() if v is not None} # Tool-result caching distinguishes "explicitly enabled" from the # field default via model_fields_set; don't let the dump turn the # default into an explicit opt-in on the copy. An agent that opted # in at construction via an explicit cache_handler (excluded from # the dump) must stay opted in — carry the consent as cache=True so # the copy wires its own fresh handler. A handler merely offered by # a crew at kickoff is runtime wiring, not consent, and must not # opt the copy in; _constructor_cache_opt_in is recorded before any # crew wiring can happen. if "cache" not in self.model_fields_set: copied_data.pop("cache", None) if self._constructor_cache_opt_in: copied_data["cache"] = True return type(self)( **copied_data, llm=existing_llm, tools=self.tools, knowledge_sources=existing_knowledge_sources, knowledge=copied_knowledge, knowledge_storage=copied_knowledge_storage, ) def interpolate_inputs(self, inputs: dict[str, Any]) -> None: """Interpolate inputs into the agent description and backstory.""" if self._original_role is None: self._original_role = self.role if self._original_goal is None: self._original_goal = self.goal if self._original_backstory is None: self._original_backstory = self.backstory if inputs: self.role = interpolate_only( input_string=self._original_role, inputs=inputs ) self.goal = interpolate_only( input_string=self._original_goal, inputs=inputs ) self.backstory = interpolate_only( input_string=self._original_backstory, inputs=inputs ) def set_cache_handler(self, cache_handler: CacheHandler) -> None: """Set the cache handler for the agent. Args: cache_handler: An instance of the CacheHandler class. """ self.tools_handler = ToolsHandler() if self.cache: self.cache_handler = cache_handler self.tools_handler.cache = cache_handler def set_rpm_controller(self, rpm_controller: RPMController) -> None: """Set the rpm controller for the agent. Args: rpm_controller: An instance of the RPMController class. """ if not self._rpm_controller: self._rpm_controller = rpm_controller def set_knowledge(self, crew_embedder: EmbedderConfig | None = None) -> None: pass def set_skills(self, resolved_crew_skills: list[Any] | None = None) -> None: pass