Centralized agent loading with proper isolation, caching, and .env loading. Support loading agents from below folder/file structures: a) {agent_name}.agent as a module name: agents_dir/{agent_name}/agent.py (with root_agent defined in the module) b) {agent_name} as a module name
| 56 | |
| 57 | |
| 58 | class AgentLoader(BaseAgentLoader): |
| 59 | """Centralized agent loading with proper isolation, caching, and .env loading. |
| 60 | Support loading agents from below folder/file structures: |
| 61 | a) {agent_name}.agent as a module name: |
| 62 | agents_dir/{agent_name}/agent.py (with root_agent defined in the module) |
| 63 | b) {agent_name} as a module name |
| 64 | agents_dir/{agent_name}.py (with root_agent defined in the module) |
| 65 | c) {agent_name} as a package name |
| 66 | agents_dir/{agent_name}/__init__.py (with root_agent in the package) |
| 67 | d) {agent_name} as a YAML config folder: |
| 68 | agents_dir/{agent_name}/root_agent.yaml defines the root agent |
| 69 | |
| 70 | """ |
| 71 | |
| 72 | def __init__(self, agents_dir: str): |
| 73 | agents_path = Path(agents_dir).resolve() |
| 74 | is_single_agent = is_single_agent_directory(agents_path) |
| 75 | if is_single_agent: |
| 76 | self._is_single_agent = True |
| 77 | self._single_agent_name = agents_path.name |
| 78 | self.agents_dir = str(agents_path.parent) |
| 79 | else: |
| 80 | self._is_single_agent = False |
| 81 | self._single_agent_name = None |
| 82 | self.agents_dir = str(agents_path) |
| 83 | |
| 84 | self._original_sys_path = None |
| 85 | self._agent_cache: dict[str, Union[BaseAgent, App]] = {} |
| 86 | |
| 87 | @property |
| 88 | def is_single_agent(self) -> bool: |
| 89 | """Returns True if the loader is in single agent mode.""" |
| 90 | return self._is_single_agent |
| 91 | |
| 92 | @property |
| 93 | def single_agent_name(self) -> Optional[str]: |
| 94 | """Returns the name of the agent in single agent mode.""" |
| 95 | return self._single_agent_name |
| 96 | |
| 97 | def _set_single_agent_mode(self, name: str, agents_dir: str) -> None: |
| 98 | """Internal method to force single agent mode. Use with care.""" |
| 99 | self._is_single_agent = True |
| 100 | self._single_agent_name = name |
| 101 | self.agents_dir = agents_dir |
| 102 | |
| 103 | def _load_from_module_or_package( |
| 104 | self, agent_name: str |
| 105 | ) -> Optional[Union[BaseAgent, App]]: |
| 106 | # Load for case: Import "{agent_name}" (as a package or module) |
| 107 | # Covers structures: |
| 108 | # a) agents_dir/{agent_name}.py (with root_agent in the module) |
| 109 | # b) agents_dir/{agent_name}/__init__.py (with root_agent in the package) |
| 110 | try: |
| 111 | module_candidate = importlib.import_module(agent_name) |
| 112 | # Check for "app" first, then "root_agent" |
| 113 | if hasattr(module_candidate, "app") and isinstance( |
| 114 | module_candidate.app, App |
| 115 | ): |
no outgoing calls