Internal logic to load an agent
(self, agent_name: str)
| 263 | ) |
| 264 | |
| 265 | def _perform_load(self, agent_name: str) -> Union[BaseAgent, App]: |
| 266 | """Internal logic to load an agent""" |
| 267 | self._validate_agent_name(agent_name) |
| 268 | # Determine the directory to use for loading |
| 269 | if agent_name.startswith("__"): |
| 270 | # Special agent: use special agents directory |
| 271 | agents_dir = os.path.abspath(SPECIAL_AGENTS_DIR) |
| 272 | # Remove the double underscore prefix for the actual agent name |
| 273 | actual_agent_name = agent_name[2:] |
| 274 | # If this special agents directory is part of a package (has __init__.py |
| 275 | # up the tree), build a fully-qualified module path so the built-in agent |
| 276 | # can continue to use relative imports. Otherwise, fall back to importing |
| 277 | # by module name relative to agents_dir. |
| 278 | module_base_name = actual_agent_name |
| 279 | package_parts: list[str] = [] |
| 280 | package_root: Optional[Path] = None |
| 281 | current_dir = Path(agents_dir).resolve() |
| 282 | while True: |
| 283 | if not (current_dir / "__init__.py").is_file(): |
| 284 | package_root = current_dir |
| 285 | break |
| 286 | package_parts.append(current_dir.name) |
| 287 | current_dir = current_dir.parent |
| 288 | if package_parts: |
| 289 | package_parts.reverse() |
| 290 | module_base_name = ".".join(package_parts + [actual_agent_name]) |
| 291 | if str(package_root) not in sys.path: |
| 292 | sys.path.insert(0, str(package_root)) |
| 293 | else: |
| 294 | # Regular agent: use the configured agents directory |
| 295 | agents_dir = self.agents_dir |
| 296 | actual_agent_name = agent_name |
| 297 | module_base_name = actual_agent_name |
| 298 | |
| 299 | # Add agents_dir to sys.path |
| 300 | if agents_dir not in sys.path: |
| 301 | sys.path.insert(0, agents_dir) |
| 302 | |
| 303 | logger.debug("Loading .env for agent %s from %s", agent_name, agents_dir) |
| 304 | envs.load_dotenv_for_agent(actual_agent_name, str(agents_dir)) |
| 305 | |
| 306 | if root_agent := self._load_from_module_or_package(module_base_name): |
| 307 | self._record_origin_metadata( |
| 308 | loaded=root_agent, |
| 309 | expected_app_name=agent_name, |
| 310 | module_name=module_base_name, |
| 311 | agents_dir=agents_dir, |
| 312 | ) |
| 313 | return root_agent |
| 314 | |
| 315 | if root_agent := self._load_from_submodule(module_base_name): |
| 316 | self._record_origin_metadata( |
| 317 | loaded=root_agent, |
| 318 | expected_app_name=agent_name, |
| 319 | module_name=f"{module_base_name}.agent", |
| 320 | agents_dir=agents_dir, |
| 321 | ) |
| 322 | return root_agent |
no test coverage detected