diff --git a/sample_agent.py b/sample_agent.py new file mode 100644 index 0000000000..a668a12696 --- /dev/null +++ b/sample_agent.py @@ -0,0 +1,61 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from google.adk.agents.llm_agent import LlmAgent +from google.adk.agents.sequential_agent import SequentialAgent + + +def search_web(query: str) -> str: + """Searches the web for given query.""" + return f"Search results for: {query}" + + +def calculate(expression: str) -> str: + """Calculates math expression.""" + return f"Calculated: {expression}" + + +# 1. Researcher Agent with web search tool +researcher = LlmAgent( + name="ResearcherAgent", + instruction="Search the web and gather background information.", + tools=[search_web], +) + +# 2. Analyst Agent with calculator tool +analyst = LlmAgent( + name="AnalystAgent", + instruction="Analyze research data and calculate statistics.", + tools=[calculate], +) + +# 3. Writer Agent +writer = LlmAgent( + name="WriterAgent", + instruction="Synthesize findings and draft final executive report.", +) + +# Root Sequential Pipeline +root_agent = SequentialAgent( + name="ResearchAndReportingPipeline", + description="Multi-agent workflow that researches, analyzes, and drafts reports.", + sub_agents=[researcher, analyst, writer], +) + +# New Sequential Pipeline +ResearchAndReportingPipeline = SequentialAgent( + name="ResearchAndReportingPipeline", + description="Multi-agent workflow that researches, analyzes, and drafts reports.", + sub_agents=[], +) diff --git a/src/google/adk/cli/cli_graph.py b/src/google/adk/cli/cli_graph.py new file mode 100644 index 0000000000..f992d98520 --- /dev/null +++ b/src/google/adk/cli/cli_graph.py @@ -0,0 +1,53 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Optional + +import click + +from .utils.agent_loader import AgentLoader +from .graph.inspector import AgentInspector +from .graph.graph_server import GraphServer + +logger = logging.getLogger("google_adk." + __name__) + + +@click.command("graph") +@click.argument("agent_file", type=click.Path(exists=True), required=False) +@click.option("--host", default="0.0.0.0", help="Host address to bind the web server.") +@click.option("--port", default=8000, type=int, help="Port to serve the visual graph UI.") +def graph_cmd(agent_file: Optional[str], host: str, port: int) -> None: + """Inspect and visualize agent topology interactively.""" + if not agent_file: + click.echo("Starting ADK Graph Server in standalone builder mode...") + server = GraphServer(topology=None, host=host, port=port) + server.run() + return + + click.echo(f"Inspecting agent at: {agent_file}") + path = Path(agent_file) + agent_or_app = AgentLoader.load_agent_or_app(path) + + inspector = AgentInspector(agent_or_app) + topology = inspector.inspect() + + click.echo(f"Successfully parsed agent graph! Total nodes: {len(topology.nodes)}, edges: {len(topology.edges)}") + click.echo(f"Serving Visual Agent Graph on http://{host}:{port}") + + server = GraphServer(topology=topology, agent_file_path=path, host=host, port=port) + server.run() diff --git a/src/google/adk/cli/cli_tools_click.py b/src/google/adk/cli/cli_tools_click.py index f14c74f9b9..cf9f48e779 100644 --- a/src/google/adk/cli/cli_tools_click.py +++ b/src/google/adk/cli/cli_tools_click.py @@ -1969,6 +1969,46 @@ def _check_windows_reload(reload: bool) -> bool: return reload +@main.command("graph") +@click.argument("agent_file", type=click.Path(exists=True), required=False) +@click.option("--host", default="127.0.0.1", help="Host address to bind the web server.") +@click.option("--port", default=8000, type=int, help="Port to serve the visual graph UI.") +def cli_graph(agent_file: Optional[str], host: str, port: int) -> None: + """Inspect and visualize agent topology interactively.""" + from .graph.graph_server import GraphServer + from .graph.inspector import AgentInspector + from .utils.agent_loader import AgentLoader + + if not agent_file: + click.echo("Starting ADK Graph Server in standalone builder mode...") + server = GraphServer(topology=None, host=host, port=port) + server.run() + return + + click.echo(f"Inspecting agent at: {agent_file}") + path = Path(agent_file).resolve() + if path.is_file(): + agents_dir = str(path.parent) + agent_name = path.stem + else: + agents_dir = str(path) + agent_name = path.name + + loader = AgentLoader(agents_dir=agents_dir) + agent_or_app = loader.load_agent(agent_name) + + inspector = AgentInspector(agent_or_app) + topology = inspector.inspect() + + click.echo(f"Successfully parsed agent graph! Total nodes: {len(topology.nodes)}, edges: {len(topology.edges)}") + click.echo(f"Serving Visual Agent Graph on http://{host}:{port}") + + server = GraphServer( + topology=topology, agent_file_path=path, host=host, port=port + ) + server.run() + + @main.command("web") @feature_options() @fast_api_common_options() diff --git a/src/google/adk/cli/graph/__init__.py b/src/google/adk/cli/graph/__init__.py new file mode 100644 index 0000000000..ac22f47559 --- /dev/null +++ b/src/google/adk/cli/graph/__init__.py @@ -0,0 +1,22 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Graph visualization and inspection package for ADK agents.""" + +from __future__ import annotations + +from .inspector import AgentInspector, GraphTopology, GraphNode, GraphEdge +from .graph_server import GraphServer + +__all__ = ["AgentInspector", "GraphTopology", "GraphNode", "GraphEdge", "GraphServer"] diff --git a/src/google/adk/cli/graph/_graph_document.py b/src/google/adk/cli/graph/_graph_document.py new file mode 100644 index 0000000000..760ef4be9e --- /dev/null +++ b/src/google/adk/cli/graph/_graph_document.py @@ -0,0 +1,354 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Validated, persisted graph documents for the visual graph workbench.""" + +from __future__ import annotations + +import ast +import hashlib +from pathlib import Path +import tempfile + +from pydantic import BaseModel +from pydantic import Field + +from .inspector import GraphEdge +from .inspector import GraphNode +from .inspector import GraphTopology + +_EDITABLE_AGENT_TYPES = frozenset( + {"llm_agent", "sequential", "parallel", "loop"} +) +_CONTAINER_TYPES = frozenset({"llm_agent", "sequential", "parallel", "loop"}) + + +class GraphDocument(BaseModel): + """A durable graph draft with optimistic-concurrency metadata.""" + + schema_version: int = 1 + revision: int = Field(default=0, ge=0) + source_digest: str | None = None + topology: GraphTopology + positions: dict[str, tuple[float, float]] = Field(default_factory=dict) + + +class GraphDocumentConflictError(RuntimeError): + """Raised when a client attempts to overwrite a newer graph draft.""" + + +class GraphDocumentStore: + """Persists graph drafts beside an agent file with revision checking.""" + + def __init__(self, *, agent_file_path: Path | None): + self._agent_file_path = agent_file_path + self._draft_path = ( + agent_file_path.parent / ".adk" / "graph-draft.json" + if agent_file_path + else None + ) + + def load(self, *, topology: GraphTopology) -> GraphDocument: + """Loads a compatible draft or initializes one from inspected topology.""" + source_digest = self._source_digest() + if not self._draft_path or not self._draft_path.is_file(): + return GraphDocument(topology=topology, source_digest=source_digest) + + try: + document = GraphDocument.model_validate_json(self._draft_path.read_text()) + except (OSError, ValueError): + return GraphDocument(topology=topology, source_digest=source_digest) + + if document.source_digest != source_digest: + return GraphDocument(topology=topology, source_digest=source_digest) + # An empty persisted draft cannot describe a non-empty source graph. This + # can happen after an interrupted first save; retaining it would leave the + # workbench permanently blank and make Preview generate a fake root agent. + if topology.nodes and not document.topology.nodes: + return GraphDocument(topology=topology, source_digest=source_digest) + return document + + def save( + self, + *, + document: GraphDocument, + expected_revision: int, + ) -> GraphDocument: + """Atomically saves a newer revision of a valid document.""" + current = self.load(topology=document.topology) + if current.revision != expected_revision: + raise GraphDocumentConflictError( + "This graph has changed in another browser. Reload before saving." + ) + + saved = document.model_copy( + update={ + "revision": expected_revision + 1, + "source_digest": self._source_digest(), + } + ) + self._write_document(saved) + return saved + + def update_source_digest(self, *, document: GraphDocument) -> GraphDocument: + """Records a successful source write without changing the draft revision.""" + updated = document.model_copy( + update={"source_digest": self._source_digest()} + ) + self._write_document(updated) + return updated + + def _write_document(self, document: GraphDocument) -> None: + if not self._draft_path: + return + self._draft_path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=self._draft_path.parent, + delete=False, + ) as temporary_file: + temporary_file.write(document.model_dump_json(indent=2)) + temporary_path = Path(temporary_file.name) + temporary_path.replace(self._draft_path) + + def _source_digest(self) -> str | None: + if not self._agent_file_path or not self._agent_file_path.is_file(): + return None + return hashlib.sha256(self._agent_file_path.read_bytes()).hexdigest() + + +def validate_topology(topology: GraphTopology) -> None: + """Ensures a graph can be represented by the managed ADK source format.""" + nodes = {node.id: node for node in topology.nodes} + if len(nodes) != len(topology.nodes): + raise ValueError("Each graph node must have a unique id.") + if topology.nodes and topology.root_id not in nodes: + raise ValueError("root_id must identify a graph node.") + if not topology.nodes and topology.root_id: + raise ValueError("An empty graph must not have a root_id.") + + parent_counts: dict[str, int] = {node_id: 0 for node_id in nodes} + tool_counts: dict[str, int] = {node_id: 0 for node_id in nodes} + adjacency: dict[str, list[str]] = {node_id: [] for node_id in nodes} + for edge in topology.edges: + _validate_edge(edge=edge, nodes=nodes) + if edge.type == "sub_agent": + parent_counts[edge.target] += 1 + adjacency[edge.source].append(edge.target) + else: + tool_counts[edge.source] += 1 + + for node in nodes.values(): + if node.type == "tool": + if tool_counts[node.id] != 1: + raise ValueError( + f"Tool {node.label!r} must be bound to exactly one LLM agent." + ) + continue + if node.type not in _EDITABLE_AGENT_TYPES: + raise ValueError( + f"{node.label!r} uses unsupported type {node.type!r}. " + "Custom agents are read-only in the graph editor." + ) + if node.id == topology.root_id: + if parent_counts[node.id]: + raise ValueError("The root agent cannot be a sub-agent.") + elif parent_counts[node.id] != 1: + raise ValueError( + f"Agent {node.label!r} must have exactly one parent connection." + ) + + _validate_acyclic(adjacency) + + +def generate_source(topology: GraphTopology) -> str: + """Generates syntactically validated Python for a managed ADK graph.""" + validate_topology(topology) + if not topology.nodes: + return ( + "from google.adk.agents import LlmAgent\n\nroot_agent =" + " LlmAgent(name='root')\n" + ) + + nodes = {node.id: node for node in topology.nodes} + children: dict[str, list[str]] = {node_id: [] for node_id in nodes} + tools: dict[str, list[str]] = {node_id: [] for node_id in nodes} + for edge in topology.edges: + if edge.type == "sub_agent": + children[edge.source].append(edge.target) + else: + tools[edge.target].append(edge.source) + + lines = [ + "from google.adk.agents import LlmAgent", + "from google.adk.agents import LoopAgent", + "from google.adk.agents import ParallelAgent", + "from google.adk.agents import SequentialAgent", + "", + ] + for node in nodes.values(): + if node.type == "tool": + lines.extend(_tool_definition(node)) + + for node_id in reversed(_agent_order(topology)): + node = nodes[node_id] + lines.extend( + _agent_definition( + node=node, + children=[nodes[child] for child in children[node_id]], + tools=[nodes[tool] for tool in tools[node_id]], + ) + ) + lines.append(f"root_agent = {_identifier(nodes[topology.root_id])}") + source = "\n".join(lines) + "\n" + ast.parse(source) + return source + + +def _validate_edge(*, edge: GraphEdge, nodes: dict[str, GraphNode]) -> None: + if edge.source not in nodes or edge.target not in nodes: + raise ValueError(f"Connection {edge.id!r} references a missing node.") + source = nodes[edge.source] + target = nodes[edge.target] + if edge.type == "sub_agent": + if source.type not in _CONTAINER_TYPES or target.type == "tool": + raise ValueError("A sub-agent connection must link an agent to an agent.") + elif edge.type == "tool_binding": + if source.type != "tool" or target.type != "llm_agent": + raise ValueError("A tool connection must link a tool to an LLM agent.") + else: + raise ValueError(f"Unsupported connection type {edge.type!r}.") + + +def _validate_acyclic(adjacency: dict[str, list[str]]) -> None: + visiting: set[str] = set() + visited: set[str] = set() + + def visit(node_id: str) -> None: + if node_id in visiting: + raise ValueError("Sub-agent connections cannot contain a cycle.") + if node_id in visited: + return + visiting.add(node_id) + for child_id in adjacency[node_id]: + visit(child_id) + visiting.remove(node_id) + visited.add(node_id) + + for node_id in adjacency: + visit(node_id) + + +def _agent_order(topology: GraphTopology) -> list[str]: + children: dict[str, list[str]] = {node.id: [] for node in topology.nodes} + for edge in topology.edges: + if edge.type == "sub_agent": + children[edge.source].append(edge.target) + order: list[str] = [] + + def visit(node_id: str) -> None: + order.append(node_id) + for child_id in children[node_id]: + visit(child_id) + + if topology.root_id: + visit(topology.root_id) + return order + + +def _agent_definition( + *, node: GraphNode, children: list[GraphNode], tools: list[GraphNode] +) -> list[str]: + identifier = _identifier(node) + description = node.description or "" + if node.type == "llm_agent": + kwargs = [f"name={node.label!r}", f"description={description!r}"] + instruction = node.config.get("instruction", "") + if isinstance(instruction, str) and instruction: + kwargs.append(f"instruction={instruction!r}") + model = node.config.get("model", "") + if isinstance(model, str) and model: + kwargs.append(f"model={model!r}") + if children: + kwargs.append( + "sub_agents=[" + + ", ".join(_identifier(child) for child in children) + + "]" + ) + if tools: + kwargs.append( + "tools=[" + ", ".join(_identifier(tool) for tool in tools) + "]" + ) + return [ + f"{identifier} = LlmAgent(", + *[f" {arg}," for arg in kwargs], + ")", + "", + ] + + class_name = { + "sequential": "SequentialAgent", + "parallel": "ParallelAgent", + "loop": "LoopAgent", + }[node.type] + kwargs = [f"name={node.label!r}", f"description={description!r}"] + kwargs.append( + "sub_agents=[" + ", ".join(_identifier(child) for child in children) + "]" + ) + if node.type == "loop": + max_iterations = node.config.get("max_iterations") + if isinstance(max_iterations, int) and max_iterations > 0: + kwargs.append(f"max_iterations={max_iterations}") + return [ + f"{identifier} = {class_name}(", + *[f" {arg}," for arg in kwargs], + ")", + "", + ] + + +def _tool_definition(node: GraphNode) -> list[str]: + identifier = _identifier(node) + implementation = node.config.get("implementation") + if not isinstance(implementation, str) or not implementation.strip(): + raise ValueError( + f"Tool {node.label!r} needs a Python implementation before source can" + " be generated." + ) + try: + parsed = ast.parse(implementation) + except SyntaxError as error: + raise ValueError( + f"Tool {node.label!r} has invalid Python: {error.msg} (line" + f" {error.lineno})." + ) from error + if not any( + isinstance(statement, (ast.AsyncFunctionDef, ast.FunctionDef)) + and statement.name == identifier + for statement in parsed.body + ): + raise ValueError( + f"Tool implementation must define a function named {identifier!r}." + ) + return [implementation.rstrip(), ""] + + +def _identifier(node: GraphNode) -> str: + if not node.label.isidentifier() or node.label == "user": + raise ValueError( + f"Node name {node.label!r} must be a Python identifier other than" + " 'user'." + ) + return node.label diff --git a/src/google/adk/cli/graph/graph_server.py b/src/google/adk/cli/graph/graph_server.py new file mode 100644 index 0000000000..d0d7205ed9 --- /dev/null +++ b/src/google/adk/cli/graph/graph_server.py @@ -0,0 +1,278 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import logging +from pathlib import Path +import tempfile +from typing import Optional + +from fastapi import FastAPI +from fastapi import HTTPException +from fastapi.responses import HTMLResponse +from pydantic import BaseModel +import uvicorn + +from ._graph_document import generate_source +from ._graph_document import GraphDocument +from ._graph_document import GraphDocumentConflictError +from ._graph_document import GraphDocumentStore +from ._graph_document import validate_topology +from .inspector import GraphTopology + +logger = logging.getLogger("google_adk." + __name__) + + +def _atomic_write(path: Path, contents: str) -> None: + """Atomically replaces a UTF-8 source file after caller-side validation.""" + if path.is_file(): + backup_path = path.parent / ".adk" / "graph-backups" / path.name + backup_path.parent.mkdir(parents=True, exist_ok=True) + backup_path.write_bytes(path.read_bytes()) + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=path.parent, delete=False + ) as temporary_file: + temporary_file.write(contents) + temporary_path = Path(temporary_file.name) + temporary_path.replace(path) + + +class CodeSaveRequest(BaseModel): + code: str + + +class GraphDraftRequest(BaseModel): + """A revision-checked graph document update from the workbench.""" + + document: GraphDocument + expected_revision: int + + +class GraphServer: + """Serves an editable visual graph workbench for ADK agents.""" + + def __init__( + self, + topology: Optional[GraphTopology], + agent_file_path: Optional[Path] = None, + host: str = "127.0.0.1", + port: int = 8000, + ): + self.agent_file_path = agent_file_path + self.host = host + self.port = port + initial_topology = ( + topology.model_copy(deep=True) + if topology + else GraphTopology(root_id="") + ) + self._document_store = GraphDocumentStore(agent_file_path=agent_file_path) + self.document = self._document_store.load(topology=initial_topology) + self.app = FastAPI(title="ADK Graph Workbench") + self._setup_routes() + + def _setup_routes(self) -> None: + @self.app.get("/api/graph/topology") + def get_topology() -> dict: + return self.document.model_dump() + + @self.app.put("/api/graph/topology") + def save_topology(req: GraphDraftRequest) -> dict: + try: + if req.expected_revision != self.document.revision: + raise GraphDocumentConflictError( + "This graph has changed in another browser. Reload before saving." + ) + validate_topology(req.document.topology) + self.document = self._document_store.save( + document=req.document, + expected_revision=req.expected_revision, + ) + except GraphDocumentConflictError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + except ValueError as error: + raise HTTPException(status_code=422, detail=str(error)) from error + return self.document.model_dump() + + @self.app.get("/api/graph/code") + def get_code() -> dict: + if not self.agent_file_path or not self.agent_file_path.exists(): + return { + "code": "# Standalone builder mode - no file selected", + "editable": False, + } + return { + "code": self.agent_file_path.read_text(encoding="utf-8"), + "editable": True, + } + + @self.app.post("/api/graph/code") + def save_code(req: CodeSaveRequest) -> dict: + if not self.agent_file_path: + raise HTTPException( + status_code=400, detail="No source file attached to save code." + ) + try: + compile(req.code, str(self.agent_file_path), "exec") + except SyntaxError as error: + raise HTTPException( + status_code=422, + detail=( + f"Source contains invalid Python: {error.msg} (line" + f" {error.lineno})." + ), + ) from error + _atomic_write(self.agent_file_path, req.code) + return { + "status": "saved", + "path": str(self.agent_file_path), + "requires_reload": True, + } + + @self.app.post("/api/graph/code/preview") + def preview_generated_code() -> dict: + try: + return {"code": generate_source(self.document.topology)} + except ValueError as error: + raise HTTPException(status_code=422, detail=str(error)) from error + + @self.app.post("/api/graph/code/apply") + def apply_generated_code() -> dict: + if not self.agent_file_path: + raise HTTPException( + status_code=400, detail="No source file attached to save code." + ) + try: + source = generate_source(self.document.topology) + except ValueError as error: + raise HTTPException(status_code=422, detail=str(error)) from error + _atomic_write(self.agent_file_path, source) + self.document = self._document_store.update_source_digest( + document=self.document + ) + return { + "status": "saved", + "path": str(self.agent_file_path), + "code": source, + } + + @self.app.get("/", response_class=HTMLResponse) + def index() -> str: + return r""" + + + + + + ADK Graph Workbench + + + + +
Google ADKGraph workbenchLoading graph…
+
+ +
100%

Start building your agent system

Add an agent or tool from the library. Your source code stays tucked away until you open the Code drawer.

+ +
+
Python source
+
+ + + + """ + + def run(self) -> None: + uvicorn.run(self.app, host=self.host, port=self.port) diff --git a/src/google/adk/cli/graph/inspector.py b/src/google/adk/cli/graph/inspector.py new file mode 100644 index 0000000000..e5e1163045 --- /dev/null +++ b/src/google/adk/cli/graph/inspector.py @@ -0,0 +1,275 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import inspect +import textwrap +from typing import Any +from typing import Dict +from typing import List +from typing import Optional +from typing import Union + +from pydantic import BaseModel +from pydantic import Field + +from ...agents.base_agent import BaseAgent +from ...agents.llm_agent import LlmAgent +from ...agents.loop_agent import LoopAgent +from ...agents.parallel_agent import ParallelAgent +from ...agents.sequential_agent import SequentialAgent +from ...apps.app import App +from ...tools.base_tool import BaseTool +from ...workflow._workflow import Workflow + + +class GraphNode(BaseModel): + id: str + type: str # Agent, tool, plugin, workflow, or custom component type. + label: str + description: Optional[str] = None + sub_agents: List[str] = Field(default_factory=list) + tools: List[str] = Field(default_factory=list) + parent_id: Optional[str] = None + config: Dict[str, Any] = Field(default_factory=dict) + + +class GraphEdge(BaseModel): + id: str + source: str + target: str + type: str # 'sub_agent', 'tool_binding', or 'app_plugin' + label: Optional[str] = None + + +class GraphTopology(BaseModel): + root_id: str + nodes: List[GraphNode] = Field(default_factory=list) + edges: List[GraphEdge] = Field(default_factory=list) + + +class AgentInspector: + """Inspects Google ADK Agent instances and converts them to a visual GraphTopology.""" + + def __init__(self, root: Union[BaseAgent, App, Any]): + self.root = root + self.nodes: Dict[str, GraphNode] = {} + self.edges: List[GraphEdge] = [] + + def inspect(self) -> GraphTopology: + if isinstance(self.root, App): + root_agent = self.root.root_agent + else: + root_agent = self.root + + root_id = self._inspect_agent(root_agent) + if isinstance(self.root, App): + for index, plugin in enumerate(self.root.plugins): + plugin_name = getattr(plugin, "name", type(plugin).__name__) + plugin_id = f"plugin_{index}_{plugin_name}" + self.nodes[plugin_id] = GraphNode( + id=plugin_id, + type="plugin", + label=plugin_name, + description=(type(plugin).__doc__ or "").strip(), + config={"class": type(plugin).__name__, "read_only": True}, + ) + self.edges.append( + GraphEdge( + id=f"edge_plugin_{plugin_id}_{root_id}", + source=plugin_id, + target=root_id, + type="app_plugin", + ) + ) + return GraphTopology( + root_id=root_id, + nodes=list(self.nodes.values()), + edges=self.edges, + ) + + def _inspect_agent(self, agent: Any, parent_id: Optional[str] = None) -> str: + if isinstance(agent, Workflow): + return self._inspect_workflow(agent, parent_id=parent_id) + agent_id = getattr(agent, "name", str(id(agent))) + + # Determine type + if isinstance(agent, SequentialAgent): + node_type = "sequential" + elif isinstance(agent, ParallelAgent): + node_type = "parallel" + elif isinstance(agent, LoopAgent): + node_type = "loop" + elif isinstance(agent, LlmAgent): + node_type = "llm_agent" + else: + node_type = "base_agent" + + config: Dict[str, Any] = {"class": type(agent).__name__} + if hasattr(agent, "model") and agent.model: + config["model"] = str(agent.model) + if hasattr(agent, "instruction") and agent.instruction: + config["instruction"] = str(agent.instruction) + if hasattr(agent, "max_iterations") and agent.max_iterations is not None: + config["max_iterations"] = agent.max_iterations + if node_type == "base_agent": + config["read_only"] = True + + node = GraphNode( + id=agent_id, + type=node_type, + label=getattr(agent, "name", "Agent"), + description=getattr(agent, "description", None), + parent_id=parent_id, + config=config, + ) + self.nodes[agent_id] = node + + # Inspect Tools + if hasattr(agent, "tools") and agent.tools: + for tool in agent.tools: + tool_id = self._inspect_tool(tool, agent_id) + node.tools.append(tool_id) + + # Inspect Sub-Agents + sub_agents = getattr(agent, "sub_agents", []) + if sub_agents: + for sub in sub_agents: + sub_id = self._inspect_agent(sub, parent_id=agent_id) + node.sub_agents.append(sub_id) + # Add edge + self.edges.append( + GraphEdge( + id=f"edge_sub_{agent_id}_{sub_id}", + source=agent_id, + target=sub_id, + type="sub_agent", + ) + ) + + return agent_id + + def _inspect_workflow( + self, workflow: Workflow, parent_id: Optional[str] = None + ) -> str: + """Exposes workflow nodes and routes without claiming they are editable.""" + workflow_id = workflow.name + self.nodes[workflow_id] = GraphNode( + id=workflow_id, + type="workflow", + label=workflow.name, + description=workflow.description, + parent_id=parent_id, + config={"class": type(workflow).__name__, "read_only": True}, + ) + if not workflow.graph: + return workflow_id + + node_ids: dict[int, str] = {} + for workflow_node in workflow.graph.nodes: + node_id = f"{workflow_id}:{workflow_node.name}" + node_ids[id(workflow_node)] = node_id + self.nodes[node_id] = GraphNode( + id=node_id, + type="workflow_node", + label=workflow_node.name, + description=workflow_node.description, + parent_id=workflow_id, + config={"class": type(workflow_node).__name__, "read_only": True}, + ) + self.edges.append( + GraphEdge( + id=f"edge_workflow_contains_{workflow_id}_{node_id}", + source=workflow_id, + target=node_id, + type="workflow_contains", + ) + ) + + for index, edge in enumerate(workflow.graph.edges): + source_id = node_ids.get(id(edge.from_node)) + target_id = node_ids.get(id(edge.to_node)) + if not source_id or not target_id: + continue + self.edges.append( + GraphEdge( + id=f"edge_workflow_route_{workflow_id}_{index}", + source=source_id, + target=target_id, + type="workflow_route", + label=str(edge.route) if edge.route is not None else None, + ) + ) + return workflow_id + + def _inspect_tool(self, tool: Union[BaseTool, Any], agent_id: str) -> str: + tool_name = getattr(tool, "name", getattr(tool, "__name__", str(id(tool)))) + tool_id = f"tool_{agent_id}_{tool_name}" + + if tool_id not in self.nodes: + doc = getattr(tool, "description", getattr(tool, "__doc__", None)) + config: Dict[str, Any] = {"class": type(tool).__name__} + implementation = self._tool_implementation(tool) + if implementation: + config["implementation"] = implementation + else: + # The graph can still show tools backed by OpenAPI, MCP, toolsets, or + # dynamically-created callables. They are intentionally read-only: + # serializing an invented implementation would make generated code lie. + config["read_only"] = True + config["generation_hint"] = ( + "This tool's implementation is not available as a Python function. " + "Add a Python implementation before generating managed source." + ) + self.nodes[tool_id] = GraphNode( + id=tool_id, + type="tool", + label=tool_name, + description=doc, + parent_id=agent_id, + config=config, + ) + + self.edges.append( + GraphEdge( + id=f"edge_tool_{tool_id}_{agent_id}", + source=tool_id, + target=agent_id, + type="tool_binding", + ) + ) + + return tool_id + + @staticmethod + def _tool_implementation(tool: Union[BaseTool, Any]) -> Optional[str]: + """Returns inspectable Python function source for a FunctionTool. + + ADK converts callables passed to ``tools`` into ``FunctionTool`` objects. + Keeping their original function source allows the graph generator to emit + a real implementation instead of a deceptive placeholder. Native tools + and dynamically-created functions are left read-only when source is not + available. + """ + function = getattr(tool, "func", tool) + if not inspect.isfunction(function): + return None + try: + source = textwrap.dedent(inspect.getsource(function)).strip() + parsed = compile(source, "", "exec") + except (OSError, TypeError, SyntaxError): + return None + del parsed + return source or None diff --git a/tests/unittests/cli/test_graph_inspector.py b/tests/unittests/cli/test_graph_inspector.py new file mode 100644 index 0000000000..c3cc4c7d85 --- /dev/null +++ b/tests/unittests/cli/test_graph_inspector.py @@ -0,0 +1,81 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from google.adk.agents.llm_agent import LlmAgent +from google.adk.agents.sequential_agent import SequentialAgent +from google.adk.apps.app import App +from google.adk.cli.graph.inspector import AgentInspector +from google.adk.plugins.base_plugin import BasePlugin +from google.adk.workflow import START +from google.adk.workflow import Workflow + + +def dummy_tool(query: str) -> str: + """Dummy tool function for testing.""" + return f"Result: {query}" + + +def test_agent_inspector_topology(): + agent_a = LlmAgent( + name="Researcher", instruction="Search web", tools=[dummy_tool] + ) + agent_b = LlmAgent(name="Writer", instruction="Write report") + + pipeline = SequentialAgent(name="Pipeline", sub_agents=[agent_a, agent_b]) + + inspector = AgentInspector(pipeline) + topology = inspector.inspect() + + assert topology.root_id == "Pipeline" + assert len(topology.nodes) == 4 # Pipeline, Researcher, Writer, dummy_tool + assert ( + len(topology.edges) == 3 + ) # Pipeline->Researcher, Pipeline->Writer, dummy_tool->Researcher + tool_node = next( + node for node in topology.nodes if node.label == "dummy_tool" + ) + assert "def dummy_tool" in tool_node.config["implementation"] + + +def test_agent_inspector_includes_application_plugins(): + """Application-wide plugins are visible as read-only graph capabilities.""" + root_agent = LlmAgent(name="Root") + plugin = BasePlugin(name="audit") + + topology = AgentInspector( + App(name="sample", root_agent=root_agent, plugins=[plugin]) + ).inspect() + + plugin_node = next(node for node in topology.nodes if node.type == "plugin") + plugin_edge = next( + edge for edge in topology.edges if edge.type == "app_plugin" + ) + assert plugin_node.label == "audit" + assert plugin_node.config["read_only"] is True + assert plugin_edge.target == "Root" + + +def test_agent_inspector_includes_workflow_nodes_and_routes(): + """Workflow scheduling nodes and route connections remain visible.""" + first = LlmAgent(name="first") + second = LlmAgent(name="second") + workflow = Workflow(name="pipeline", edges=[(START, first), (first, second)]) + + topology = AgentInspector(workflow).inspect() + + assert any(node.type == "workflow" for node in topology.nodes) + assert {"pipeline:first", "pipeline:second"}.issubset( + {node.id for node in topology.nodes} + ) + assert any(edge.type == "workflow_route" for edge in topology.edges) diff --git a/tests/unittests/cli/test_graph_server.py b/tests/unittests/cli/test_graph_server.py new file mode 100644 index 0000000000..671d5b0633 --- /dev/null +++ b/tests/unittests/cli/test_graph_server.py @@ -0,0 +1,370 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import ast +from pathlib import Path + +from fastapi.testclient import TestClient +from google.adk.cli.graph._graph_document import generate_source +from google.adk.cli.graph._graph_document import GraphDocument +from google.adk.cli.graph.graph_server import GraphServer +from google.adk.cli.graph.inspector import GraphEdge +from google.adk.cli.graph.inspector import GraphNode +from google.adk.cli.graph.inspector import GraphTopology +import pytest + + +def test_graph_draft_updates_the_topology_visible_to_the_workbench() -> None: + """Saving a draft makes newly added canvas nodes available on reload.""" + server = GraphServer(topology=GraphTopology(root_id="root")) + client = TestClient(server.app) + draft = GraphTopology( + root_id="researcher", + nodes=[ + GraphNode( + id="researcher", + type="llm_agent", + label="Researcher", + ) + ], + ) + + document = GraphDocument( + topology=draft, + positions={"researcher": (240.0, 120.0)}, + ) + + response = client.put( + "/api/graph/topology", + json={"document": document.model_dump(), "expected_revision": 0}, + ) + + assert response.status_code == 200 + assert response.json()["revision"] == 1 + assert response.json()["positions"] == {"researcher": [240.0, 120.0]} + assert ( + client.get("/api/graph/topology").json()["topology"] == draft.model_dump() + ) + + +def test_graph_server_defaults_to_loopback() -> None: + """The source-writing workbench is never network-exposed by default.""" + server = GraphServer(topology=GraphTopology(root_id="")) + + assert server.host == "127.0.0.1" + + +def test_standalone_graph_exposes_read_only_code() -> None: + """A graph without an agent source disables code editing in the workbench.""" + client = TestClient(GraphServer(topology=None).app) + + response = client.get("/api/graph/code") + + assert response.status_code == 200 + assert response.json()["editable"] is False + + +def test_workbench_exposes_smart_layout_and_accessible_icon_tooltips() -> None: + """The canvas sizes labels, groups relationships, and labels icon controls.""" + client = TestClient(GraphServer(topology=None).app) + + page = client.get("/").text + + assert "wrapText(node.label,24)" in page + assert "levelSeparation:190" in page + assert "function advancedLayout()" in page + assert "function highlightRelatedNodes(id)" in page + assert "function validateDraft()" in page + assert "function inspectEdge(id)" in page + assert 'id="node-instruction"' in page + assert 'id="reconnect-edge"' in page + assert "smooth:{type:'horizontal'" in page + assert 'id="layout" class="btn">Auto layout' in page + assert 'id="add-node" class="btn primary">+ Add' in page + assert "function toolsFor(node)" in page + assert "node.type!=='tool'" in page + assert "split(/\\s+/)" in page + assert "function undo()" in page + assert "function redo()" in page + assert "let graphDocument" in page + assert "let document =" not in page + assert "font color" not in page + + +def test_graph_rejects_a_stale_draft_save() -> None: + """Concurrent browser saves cannot silently overwrite each other.""" + client = TestClient(GraphServer(topology=GraphTopology(root_id="")).app) + document = GraphDocument(topology=GraphTopology(root_id="")) + + first_response = client.put( + "/api/graph/topology", + json={"document": document.model_dump(), "expected_revision": 0}, + ) + stale_response = client.put( + "/api/graph/topology", + json={"document": document.model_dump(), "expected_revision": 0}, + ) + + assert first_response.status_code == 200 + assert stale_response.status_code == 409 + + +def test_graph_rejects_tool_bound_to_a_non_llm_agent() -> None: + """Tools cannot be connected to workflow containers or other tools.""" + topology = GraphTopology( + root_id="pipeline", + nodes=[ + GraphNode(id="pipeline", type="sequential", label="pipeline"), + GraphNode(id="search", type="tool", label="search"), + ], + edges=[ + GraphEdge( + id="bad-tool-edge", + source="search", + target="pipeline", + type="tool_binding", + ) + ], + ) + client = TestClient(GraphServer(topology=GraphTopology(root_id="")).app) + + response = client.put( + "/api/graph/topology", + json={ + "document": GraphDocument(topology=topology).model_dump(), + "expected_revision": 0, + }, + ) + + assert response.status_code == 422 + assert "tool connection" in response.json()["detail"].lower() + + +def test_generated_source_is_valid_python_for_a_connected_agent_graph() -> None: + """A managed graph produces source with agents, children, and tools wired.""" + topology = GraphTopology( + root_id="coordinator", + nodes=[ + GraphNode( + id="coordinator", + type="llm_agent", + label="coordinator", + config={"instruction": "Delegate research."}, + ), + GraphNode( + id="researcher", + type="llm_agent", + label="researcher", + config={"model": "gemini-2.5-flash"}, + ), + GraphNode( + id="search_web", + type="tool", + label="search_web", + description="Searches the web.", + config={ + "implementation": ( + "def search_web(query: str) -> str:\n" + " return f'Results for {query}'" + ) + }, + ), + ], + edges=[ + GraphEdge( + id="child", + source="coordinator", + target="researcher", + type="sub_agent", + ), + GraphEdge( + id="tool", + source="search_web", + target="researcher", + type="tool_binding", + ), + ], + ) + + source = generate_source(topology) + + ast.parse(source) + assert "sub_agents=[researcher]" in source + assert "tools=[search_web]" in source + assert "root_agent = coordinator" in source + + +def test_generated_source_rejects_tools_without_an_implementation() -> None: + """Source generation never replaces a real tool with a nonfunctional stub.""" + topology = GraphTopology( + root_id="agent", + nodes=[ + GraphNode(id="agent", type="llm_agent", label="agent"), + GraphNode(id="lookup", type="tool", label="lookup"), + ], + edges=[ + GraphEdge( + id="tool", source="lookup", target="agent", type="tool_binding" + ) + ], + ) + + with pytest.raises(ValueError, match="needs a Python implementation"): + generate_source(topology) + + +def test_generated_source_rejects_non_python_node_names() -> None: + """Generated source cannot contain an invalid or reserved ADK agent name.""" + topology = GraphTopology( + root_id="invalid name", + nodes=[ + GraphNode(id="invalid name", type="llm_agent", label="invalid name") + ], + ) + + with pytest.raises(ValueError, match="Python identifier"): + generate_source(topology) + + +def test_generated_container_source_constructs_the_adk_agent_tree() -> None: + """Sequential containers are generated in child-first construction order.""" + topology = GraphTopology( + root_id="pipeline", + nodes=[ + GraphNode(id="pipeline", type="sequential", label="pipeline"), + GraphNode(id="writer", type="llm_agent", label="writer"), + ], + edges=[ + GraphEdge( + id="child", source="pipeline", target="writer", type="sub_agent" + ) + ], + ) + namespace: dict[str, object] = {} + + exec(compile(generate_source(topology), "", "exec"), namespace) + + assert namespace["root_agent"].name == "pipeline" + assert namespace["root_agent"].sub_agents[0].name == "writer" + + +def test_applying_a_graph_writes_valid_source_and_preserves_the_draft( + tmp_path: Path, +) -> None: + """Applying a managed graph atomically writes its generated ADK source.""" + agent_file = tmp_path / "agent.py" + agent_file.write_text("# original source\n", encoding="utf-8") + topology = GraphTopology( + root_id="researcher", + nodes=[ + GraphNode( + id="researcher", + type="llm_agent", + label="researcher", + config={"instruction": "Research the supplied question."}, + ) + ], + ) + server = GraphServer(topology=topology, agent_file_path=agent_file) + client = TestClient(server.app) + + response = client.post("/api/graph/code/apply") + + assert response.status_code == 200 + assert agent_file.read_text(encoding="utf-8") == response.json()["code"] + ast.parse(response.json()["code"]) + persisted = GraphServer( + topology=GraphTopology(root_id=""), agent_file_path=agent_file + ) + assert persisted.document.topology == topology + + +def test_preview_shows_generated_source_without_modifying_the_agent_file( + tmp_path: Path, +) -> None: + """Previewing a graph does not overwrite source before an explicit apply.""" + agent_file = tmp_path / "agent.py" + agent_file.write_text("# original source\n", encoding="utf-8") + topology = GraphTopology( + root_id="writer", + nodes=[GraphNode(id="writer", type="llm_agent", label="writer")], + ) + client = TestClient( + GraphServer(topology=topology, agent_file_path=agent_file).app + ) + + response = client.post("/api/graph/code/preview") + + assert response.status_code == 200 + assert "root_agent = writer" in response.json()["code"] + assert agent_file.read_text(encoding="utf-8") == "# original source\n" + + +def test_empty_persisted_draft_does_not_hide_an_inspected_graph( + tmp_path: Path, +) -> None: + """A partial blank draft recovers to the authoritative inspected graph.""" + agent_file = tmp_path / "agent.py" + agent_file.write_text("root_agent = None\n", encoding="utf-8") + topology = GraphTopology( + root_id="writer", + nodes=[GraphNode(id="writer", type="llm_agent", label="writer")], + ) + server = GraphServer(topology=topology, agent_file_path=agent_file) + server._document_store.save( + document=GraphDocument(topology=GraphTopology(root_id="")), + expected_revision=0, + ) + + recovered = GraphServer(topology=topology, agent_file_path=agent_file) + + assert recovered.document.topology == topology + + +def test_invalid_manual_source_is_not_written(tmp_path: Path) -> None: + """Saving malformed Python preserves the last working source file.""" + agent_file = tmp_path / "agent.py" + agent_file.write_text("root_agent = None\n", encoding="utf-8") + client = TestClient( + GraphServer( + topology=GraphTopology(root_id=""), agent_file_path=agent_file + ).app + ) + + response = client.post("/api/graph/code", json={"code": "def broken(:\n"}) + + assert response.status_code == 422 + assert agent_file.read_text(encoding="utf-8") == "root_agent = None\n" + + +def test_manual_source_save_creates_a_backup_and_requires_graph_reload( + tmp_path: Path, +) -> None: + """Manual source changes preserve recovery data and mark topology stale.""" + agent_file = tmp_path / "agent.py" + agent_file.write_text("root_agent = None\n", encoding="utf-8") + client = TestClient( + GraphServer( + topology=GraphTopology(root_id=""), agent_file_path=agent_file + ).app + ) + + response = client.post("/api/graph/code", json={"code": "root_agent = 1\n"}) + + assert response.status_code == 200 + assert response.json()["requires_reload"] is True + assert agent_file.read_text(encoding="utf-8") == "root_agent = 1\n" + assert (tmp_path / ".adk" / "graph-backups" / "agent.py").read_text( + encoding="utf-8" + ) == "root_agent = None\n" diff --git a/tests/unittests/cli/utils/test_cli_tools_click.py b/tests/unittests/cli/utils/test_cli_tools_click.py index dddda4471f..0817b06b77 100644 --- a/tests/unittests/cli/utils/test_cli_tools_click.py +++ b/tests/unittests/cli/utils/test_cli_tools_click.py @@ -111,6 +111,37 @@ def test_main_disables_click_windows_glob_expansion() -> None: assert mock_main.call_args.kwargs["windows_expand_args"] is False +def test_graph_opens_the_selected_agent_source_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The graph workbench retains the selected source file for code editing.""" + from google.adk.cli.utils.agent_loader import AgentLoader + + agent_file = tmp_path / "sample_agent.py" + agent_file.touch() + captured_kwargs = {} + + class _GraphServer: + + def __init__(self, **kwargs: Any) -> None: + captured_kwargs.update(kwargs) + + def run(self) -> None: + pass + + monkeypatch.setattr( + AgentLoader, "load_agent", lambda _self, _agent_name: root_agent + ) + monkeypatch.setattr( + "google.adk.cli.graph.graph_server.GraphServer", _GraphServer + ) + + result = CliRunner().invoke(cli_tools_click.main, ["graph", str(agent_file)]) + + assert result.exit_code == 0, result.output + assert captured_kwargs["agent_file_path"] == agent_file.resolve() + + # validate_exclusive def test_validate_exclusive_allows_single() -> None: """Providing exactly one exclusive option should pass."""