Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions src/google/adk/cli/utils/agent_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ def __init__(self, agents_dir: str):
self._init_agent_mode(agents_path)
self._original_sys_path = None
self._agent_cache: dict[str, Union[BaseAgent, App]] = {}
self._root_agent_type_mismatches: list[tuple[str, str, bool]] = []

def _init_agent_mode(self, agents_path: Path) -> None:
if is_single_agent_directory(agents_path):
Expand Down Expand Up @@ -106,6 +107,15 @@ def _set_single_agent_mode(self, name: str, agents_dir: str) -> None:
self._single_agent_name = name
self.agents_dir = agents_dir

def _record_root_agent_type_mismatch(self, location: str, value: Any) -> None:
"""Records a found `root_agent` whose type prevents loading it."""
value_type = type(value)
self._root_agent_type_mismatches.append((
location,
f"{value_type.__module__}.{value_type.__qualname__}",
isinstance(value, App),
))

def _load_from_module_or_package(
self, agent_name: str
) -> Optional[Union[BaseAgent, App]]:
Expand Down Expand Up @@ -133,6 +143,9 @@ def _load_from_module_or_package(
"Root agent found is not an instance of BaseAgent. But a type %s",
type(module_candidate.root_agent),
)
self._record_root_agent_type_mismatch(
agent_name, module_candidate.root_agent
)
else:
logger.debug(
"Module %s has no root_agent. Trying next pattern.",
Expand Down Expand Up @@ -182,6 +195,9 @@ def _load_from_submodule(
"Root agent found is not an instance of BaseAgent. But a type %s",
type(module_candidate.root_agent),
)
self._record_root_agent_type_mismatch(
f"{agent_name}.agent", module_candidate.root_agent
)
else:
logger.debug(
"Module %s.agent has no root_agent.",
Expand Down Expand Up @@ -275,6 +291,7 @@ def _validate_agent_name(self, agent_name: str) -> None:

def _perform_load(self, agent_name: str) -> Union[BaseAgent, App]:
"""Internal logic to load an agent"""
self._root_agent_type_mismatches = []
self._validate_agent_name(agent_name)
# Determine the directory to use for loading
if agent_name.startswith("__"):
Expand Down Expand Up @@ -341,6 +358,25 @@ def _perform_load(self, agent_name: str) -> Union[BaseAgent, App]:
)
return root_agent

# A root_agent was found but had an unusable type: surface that
# diagnosis instead of the generic not-found guidance (#6606).
if self._root_agent_type_mismatches:
details = "\n".join(
f" - '{location}.root_agent' is of type '{type_name}'"
for location, type_name, _ in self._root_agent_type_mismatches
)
app_hint = ""
if any(is_app for _, _, is_app in self._root_agent_type_mismatches):
app_hint = (
"\n\nHINT: To serve an App, expose it under the name 'app'"
" instead of 'root_agent'."
)
raise ValueError(
f"A 'root_agent' was found for '{agent_name}' but it is not a"
" BaseAgent (or workflow node) instance:\n"
f"{details}\n\nExpose a BaseAgent instance named"
f" 'root_agent'.{app_hint}"
)
# If no root_agent was found by any pattern
# Check if user might be in the wrong directory
hint = ""
Expand Down
96 changes: 96 additions & 0 deletions tests/unittests/cli/utils/test_agent_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -1061,6 +1061,102 @@ def test_validate_agent_name_allows_special_agents_when_enabled(self):
# Should not raise any exception
loader._validate_agent_name("__adk_agent_builder_assistant")

def test_wrong_type_root_agent_raises_targeted_error(self):
"""A non-agent `root_agent` raises a type-mismatch error, not 'not found'."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
agent_file = temp_path / "mistyped_agent.py"
agent_file.write_text(dedent("""
root_agent = "I am a string, not an agent"
"""))

loader = AgentLoader(str(temp_path))

with pytest.raises(ValueError) as exc_info:
loader.load_agent("mistyped_agent")

message = str(exc_info.value)
assert "mistyped_agent.root_agent" in message
assert "builtins.str" in message
assert "No root_agent found" not in message

def test_app_exported_as_root_agent_suggests_app_name(self):
"""Exporting an App as `root_agent` points the user at the 'app' name."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
agent_file = temp_path / "app_agent.py"
agent_file.write_text(dedent("""
from google.adk.agents.base_agent import BaseAgent
from google.adk.apps.app import App


class MyAgent(BaseAgent):

def __init__(self):
super().__init__(name="my_agent")


root_agent = App(name="app_agent", root_agent=MyAgent())
"""))

loader = AgentLoader(str(temp_path))

with pytest.raises(ValueError) as exc_info:
loader.load_agent("app_agent")

message = str(exc_info.value)
assert "google.adk.apps.app.App" in message
assert "under the name 'app'" in message

def test_wrong_type_root_agent_in_agent_module_raises_targeted_error(self):
"""A non-agent `root_agent` in {agent}/agent.py reports the submodule."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
agent_dir = temp_path / "mistyped_pkg"
agent_dir.mkdir()
(agent_dir / "__init__.py").write_text("")
(agent_dir / "agent.py").write_text(dedent("""
root_agent = 42
"""))

loader = AgentLoader(str(temp_path))

with pytest.raises(ValueError) as exc_info:
loader.load_agent("mistyped_pkg")

message = str(exc_info.value)
assert "mistyped_pkg.agent.root_agent" in message
assert "builtins.int" in message
assert "No root_agent found" not in message

def test_valid_agent_module_wins_over_mistyped_package_root_agent(self):
"""A bad `root_agent` in __init__.py must not block agent.py's valid one."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
agent_dir = temp_path / "fallthrough_agent"
agent_dir.mkdir()
(agent_dir / "__init__.py").write_text(dedent("""
root_agent = "not an agent"
"""))
(agent_dir / "agent.py").write_text(dedent("""
from google.adk.agents.base_agent import BaseAgent


class FallthroughAgent(BaseAgent):

def __init__(self):
super().__init__(name="fallthrough_agent")


root_agent = FallthroughAgent()
"""))

loader = AgentLoader(str(temp_path))

agent = loader.load_agent("fallthrough_agent")

assert agent.name == "fallthrough_agent"


class TestDetermineAgentLanguage:
"""Tests for AgentLoader._determine_agent_language covering all 4 load patterns."""
Expand Down