fix(cli): keep graph edges attached to node names with a colon

src/google/adk/cli/agent_graph.py

@@ -28,6 +28,7 @@
 from ..tools.agent_tool import AgentTool
 from ..tools.base_tool import BaseTool
 from ..tools.function_tool import FunctionTool
+from .utils.graph_visualization import _Digraph
 
 if TYPE_CHECKING:
   from collections.abc import Sequence
@@ -235,7 +236,7 @@ async def draw_node(tool_or_agent: Union[BaseAgent, BaseTool]) -> None:
         if name in highlight_tuple:
           # if in highlight, draw highlight node
           if as_cluster:
-            cluster = graphviz.Digraph(
+            cluster = _Digraph(
                 name='cluster_' + name
             )  # adding "cluster_" to the name makes the graph render as a cluster subgraph
             await build_cluster(cluster, agent, name)
@@ -253,7 +254,7 @@ async def draw_node(tool_or_agent: Union[BaseAgent, BaseTool]) -> None:
           return
     # if not in highlight, draw non-highlight node
     if as_cluster:
-      cluster = graphviz.Digraph(
+      cluster = _Digraph(
           name='cluster_' + name
       )  # adding "cluster_" to the name makes the graph render as a cluster subgraph
       await build_cluster(cluster, agent, name)
@@ -314,7 +315,7 @@ async def get_agent_graph(
     dark_mode: bool = True,
 ) -> graphviz.Digraph | bytes:
   bg_color = '#333537' if dark_mode else '#ffffff'
-  graph = graphviz.Digraph(
+  graph = _Digraph(
       graph_attr={'rankdir': 'LR', 'bgcolor': bg_color}, strict=True
   )
   await build_graph(graph, root_agent, highlights_pairs)

src/google/adk/cli/utils/graph_visualization.py

@@ -21,10 +21,27 @@
 from typing import cast
 
 import graphviz
+from graphviz import quoting
 
 from ...workflow._node_status import NodeStatus
 
 
+class _Digraph(graphviz.Digraph):  # type: ignore[misc]  # graphviz is untyped
+  """A Digraph whose edge endpoints are quoted as plain node ids.
+
+  Digraph.edge() splits an endpoint on ":" before quoting it, treating the
+  rest as a port, so an edge to a node named "srv:tool" is written as
+  `a -> srv:tool` and attaches to a phantom node "srv". Quoting the whole
+  endpoint keeps the real name as the DOT id, which adk-web reads back from
+  the SVG title as the node name. No port syntax is used in these graphs.
+
+  This overrides the private `_quote_edge` hook. Verified on graphviz 0.20.2
+  and 0.21; a unit test fails if the hook is renamed or stops working.
+  """
+
+  _quote_edge = staticmethod(quoting.quote)
+
+
 def plot_workflow_graph(
     app_info: dict[str, Any],
     agent_state: dict[str, Any] | None = None,
@@ -64,7 +81,7 @@ def _traverse_sub_agents(
     graph = {"nodes": nodes, "edges": edges}
 
   nodes_state = agent_state.get("nodes", {})
-  dot = graphviz.Digraph(comment="Workflow Visualization")
+  dot = _Digraph(comment="Workflow Visualization")
 
   if dark_mode:
     graph_bgcolor = "#0F172A"

tests/unittests/cli/test_agent_graph.py

@@ -24,6 +24,7 @@
 from google.adk.agents.sequential_agent import SequentialAgent
 from google.adk.cli.agent_graph import get_agent_graph
 from google.adk.tools.agent_tool import AgentTool
+from google.adk.tools.function_tool import FunctionTool
 import pytest
 
 _DARK_GREEN = '#0F5223'
@@ -250,3 +251,66 @@ async def test_get_agent_graph_is_strict_so_repeated_edges_collapse():
 
   assert graph.strict
   assert graph.source.count('child -> roll_dice') == 1
+
+
+def _colon_tool() -> FunctionTool:
+  """A tool whose name contains ':', which Gemini function names allow."""
+  tool = FunctionTool(roll_dice)
+  tool.name = 'srv:tool'
+  return tool
+
+
+def _assert_tool_edge_lands_on_declared_node(source: str, agent: str) -> None:
+  """Checks `agent -> "srv:tool"` targets the node, not a phantom port.
+
+  _parse() unquotes names, so it cannot tell `a -> srv:tool` (port syntax)
+  from `a -> "srv:tool"`; compare the raw tokens instead.
+  """
+  declared = {
+      match['name']
+      for line in source.splitlines()
+      if (match := _NODE_RE.match(line.strip()))
+  }
+  endpoints = {
+      match[end]
+      for line in source.splitlines()
+      if (match := _EDGE_RE.match(line.strip()))
+      for end in ('src', 'dst')
+  }
+  assert '"srv:tool"' in declared
+  assert endpoints <= declared
+  assert 'srv' not in endpoints
+  assert f'{agent} -> "srv:tool"' in source
+
+
+@pytest.mark.asyncio
+async def test_build_graph_tool_name_with_colon_is_linked_to_its_agent():
+  root = LlmAgent(name='root', model='gemini-2.0-flash', tools=[_colon_tool()])
+
+  graph = await get_agent_graph(root, [])
+
+  _assert_tool_edge_lands_on_declared_node(graph.source, 'root')
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'workflow_agent_class', [SequentialAgent, LoopAgent, ParallelAgent]
+)
+async def test_build_graph_tool_name_with_colon_is_linked_inside_a_cluster(
+    workflow_agent_class,
+):
+  workflow = workflow_agent_class(
+      name='steps',
+      sub_agents=[
+          LlmAgent(
+              name='first', model='gemini-2.0-flash', tools=[_colon_tool()]
+          ),
+          LlmAgent(name='second', model='gemini-2.0-flash'),
+      ],
+  )
+
+  graph = await get_agent_graph(workflow, [])
+
+  # The tool edge is drawn on the cluster's own Digraph.
+  assert 'subgraph "cluster_steps' in graph.source
+  _assert_tool_edge_lands_on_declared_node(graph.source, 'first')

tests/unittests/cli/utils/test_graph_visualization.py

@@ -0,0 +1,194 @@
+# 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.
+
+"""Tests for plot_workflow_graph DOT output."""
+
+import re
+from typing import Any
+
+from google.adk.cli.utils.graph_visualization import _Digraph
+from google.adk.cli.utils.graph_visualization import plot_workflow_graph
+from google.adk.workflow._node_status import NodeStatus
+import graphviz
+
+_TOKEN = r'"(?:[^"\\]|\\.)*"|[^\s"\[]+'
+_NODE_LINE = re.compile(rf'^\t({_TOKEN}) \[', re.MULTILINE)
+_EDGE_LINE = re.compile(rf'^\t({_TOKEN}) -> ({_TOKEN})', re.MULTILINE)
+
+
+def _workflow_info(
+    nodes: list[dict[str, Any]], edges: list[tuple[str, str]]
+) -> dict[str, Any]:
+  return {
+      'root_agent': {
+          'name': 'wf',
+          'graph': {
+              'nodes': nodes,
+              'edges': [
+                  {'from_node': {'name': a}, 'to_node': {'name': b}}
+                  for a, b in edges
+              ],
+          },
+      }
+  }
+
+
+def _plot(info: dict[str, Any], state: dict[str, Any] | None = None) -> str:
+  dot = plot_workflow_graph(info, state, format='dot')
+  assert isinstance(dot, str)
+  return dot
+
+
+def _node_line(dot: str, name: str) -> str:
+  return next(
+      line for line in dot.splitlines() if f' {name}>' in line and '[' in line
+  )
+
+
+def test_edges_attach_to_nodes_whose_names_contain_colons():
+  """Edges to colon-named nodes hit the real nodes, not phantom ports.
+
+  The real name must stay the DOT id because adk-web reads the SVG title as
+  the node name for click handling.
+  """
+  info = _workflow_info(
+      [
+          {'name': '__START__'},
+          {
+              'name': 'a',
+              'type': 'agent',
+              'tools': [{'name': 'srv:tool', 'type': 'tool'}],
+          },
+          {'name': 'step:two', 'type': 'function'},
+      ],
+      [('__START__', 'a'), ('a', 'step:two')],
+  )
+
+  dot = _plot(info)
+
+  declared = set(_NODE_LINE.findall(dot))
+  endpoints = {end for edge in _EDGE_LINE.findall(dot) for end in edge}
+  assert {'"srv:tool"', '"step:two"'} <= declared
+  assert endpoints <= declared
+  assert not {'srv', 'step'} & endpoints
+
+
+def _assert_colon_node_is_wired(dot: str, edge: str) -> None:
+  """Checks `edge` is present and the colon node "b:x" has no phantom port."""
+  declared = set(_NODE_LINE.findall(dot))
+  endpoints = {end for pair in _EDGE_LINE.findall(dot) for end in pair}
+  assert '"b:x"' in declared
+  assert endpoints <= declared
+  assert 'b' not in endpoints
+  assert f'\t{edge}' in dot
+
+
+def test_workflow_node_with_colon_can_be_an_edge_target():
+  """An edge into a node named "b:x" lands on the declared node."""
+  info = _workflow_info(
+      [
+          {'name': 'a', 'type': 'agent'},
+          {'name': 'b:x', 'type': 'function'},
+      ],
+      [('a', 'b:x')],
+  )
+
+  _assert_colon_node_is_wired(_plot(info), 'a -> "b:x"')
+
+
+def test_workflow_node_with_colon_can_be_an_edge_source():
+  """An edge out of a node named "b:x" starts at the declared node."""
+  info = _workflow_info(
+      [
+          {'name': 'b:x', 'type': 'function'},
+          {'name': 'c', 'type': 'agent'},
+      ],
+      [('b:x', 'c')],
+  )
+
+  _assert_colon_node_is_wired(_plot(info), '"b:x" -> c')
+
+
+def test_workflow_terminal_node_with_colon_is_linked_to_end():
+  """A terminal node named "b:x" gets its edge to END from the real node."""
+  info = _workflow_info(
+      [
+          {'name': 'a', 'type': 'agent'},
+          {'name': 'b:x', 'type': 'function'},
+      ],
+      [('a', 'b:x')],
+  )
+
+  dot = _plot(info)
+
+  _assert_colon_node_is_wired(dot, '"b:x" -> __END__')
+  assert '\t__END__ [label=END' in dot
+
+
+def test_digraph_edge_hook_quotes_colon_endpoints():
+  """The graphviz hook that keeps colon names whole still exists and works."""
+  assert hasattr(graphviz.Digraph, '_quote_edge'), (
+      'graphviz renamed Digraph._quote_edge; update _Digraph in'
+      ' graph_visualization.py'
+  )
+
+  dot = _Digraph()
+  dot.edge('a', 'srv:tool')
+
+  assert 'a -> "srv:tool"' in dot.source
+
+
+def test_start_and_end_nodes_keep_their_identifiers():
+  """START and END keep the __START__ and __END__ identifiers."""
+  info = _workflow_info(
+      [{'name': '__START__'}, {'name': 'a', 'type': 'agent'}],
+      [('__START__', 'a')],
+  )
+
+  dot = _plot(info)
+
+  assert '\t__START__ [label=START' in dot
+  assert '\t__START__ -> a' in dot
+  assert '\t__END__ [label=END' in dot
+  assert '\ta -> __END__' in dot
+
+
+def test_tools_are_injected_as_dashed_nodes_with_edges_from_their_agent():
+  """A node's tools show up as dashed tool nodes linked from the node."""
+  info = _workflow_info(
+      [{'name': 'a', 'type': 'agent', 'tools': [{'name': 'search'}]}], []
+  )
+
+  dot = _plot(info)
+
+  assert 'rounded,filled,dashed' in _node_line(dot, 'search')
+  assert '\ta -> search [' in dot
+  assert 'style=dashed' in dot
+
+
+def test_completed_node_is_filled_with_completed_status_color():
+  """A completed node is filled with the completed color, others are not."""
+  info = _workflow_info(
+      [
+          {'name': 'a', 'type': 'agent'},
+          {'name': 'b', 'type': 'agent'},
+      ],
+      [('a', 'b')],
+  )
+  state = {'nodes': {'a': {'status': NodeStatus.COMPLETED.value}}}
+
+  dot = _plot(info, state)
+
+  assert 'fillcolor="#16A34A"' in _node_line(dot, 'a')
+  assert 'fillcolor="#16A34A"' not in _node_line(dot, 'b')