class GraphTool(AXMTool):
"""Import dependency graph with text/mermaid/json output.
Registered as ``ast_graph`` via axm.tools entry point.
Workspace-aware: if path is a uv workspace root, returns
inter-package dependency graph.
"""
@property
def name(self) -> str:
"""Return tool name for registry lookup."""
return "ast_graph"
@safe_execute
def execute(
self,
*,
path: str = ".",
format: str = "json",
scope: str | None = None,
**kwargs: object,
) -> ToolResult:
"""Generate import dependency graph.
Args:
path: Path to package or workspace directory.
format: Output format — 'json', 'mermaid', or 'text'.
scope: Graph scope. ``None`` (default) keeps today's
auto-detection (workspace root → package-level dep
graph; package dir → intra-package module graph).
``"package"`` forces the intra-package module graph,
``"workspace"`` builds the merged module-level workspace
graph (namespaced ``{pkg}.{module}`` nodes), and
``"workspace-deps"`` forces the package-level dependency
graph.
Returns:
ToolResult with graph data.
"""
project_path = Path(path).resolve()
try:
if scope is not None:
return self._execute_scoped(project_path, scope=scope, format=format)
# scope is None — preserve legacy auto-detection.
if self._detect_workspace(project_path):
return self._execute_workspace(project_path, format=format)
return self._execute_package_checked(project_path, format=format)
except Exception as exc: # noqa: BLE001
return ToolResult(success=False, error=str(exc))
def _execute_scoped(
self, project_path: Path, *, scope: str, format: str
) -> ToolResult:
"""Dispatch an explicit ``scope`` to its executor."""
if scope == "workspace":
return self._execute_workspace_modules(project_path, format=format)
if scope == "workspace-deps":
return self._execute_workspace(project_path, format=format)
return self._execute_package_checked(project_path, format=format)
def _execute_package_checked(
self, project_path: Path, *, format: str
) -> ToolResult:
"""Run the package executor, guarding against non-directory paths."""
if not project_path.is_dir():
return ToolResult(success=False, error=f"Not a directory: {project_path}")
return self._execute_package(project_path, format=format)
def _execute_workspace_modules(
self, project_path: Path, *, format: str
) -> ToolResult:
"""Build the merged module-level workspace graph (namespaced nodes)."""
from axm_ast.core.workspace import (
analyze_workspace,
build_workspace_module_graph,
)
ws = analyze_workspace(project_path)
graph = build_workspace_module_graph(ws)
nodes = sorted(set(graph) | {t for targets in graph.values() for t in targets})
ws_name = project_path.name
data: dict[str, object] = {"graph": graph, "nodes": nodes}
text = self.render_pkg_text(
pkg_name=f"{ws_name} workspace",
nodes=nodes,
graph=graph,
mermaid_str=None,
)
if format == "mermaid":
mermaid_str = self._modules_mermaid(graph)
data["mermaid"] = mermaid_str
text = self.render_pkg_text(
pkg_name=f"{ws_name} workspace",
nodes=nodes,
graph=graph,
mermaid_str=mermaid_str,
)
elif format == "text":
data["text"] = self._format_text(nodes, graph)
return ToolResult(success=True, data=data, text=text)
@staticmethod
def _modules_mermaid(graph: dict[str, list[str]]) -> str:
"""Render a module-level graph as a mermaid flowchart."""
nodes = sorted(set(graph) | {t for targets in graph.values() for t in targets})
ids = {name: f"n{i}" for i, name in enumerate(nodes)}
lines = ["graph TD"]
for name in nodes:
lines.append(f' {ids[name]}["{name}"]')
for src, targets in graph.items():
for target in targets:
lines.append(f" {ids[src]} --> {ids[target]}")
return "\n".join(lines)
def _detect_workspace(self, project_path: Path) -> bool:
"""Return True if *project_path* is a uv workspace root."""
from axm_ast.core.workspace import detect_workspace
return detect_workspace(project_path) is not None
def _execute_workspace(self, project_path: Path, *, format: str) -> ToolResult:
"""Build inter-package dependency graph for a uv workspace."""
from axm_ast.core.workspace import (
analyze_workspace,
build_workspace_dep_graph,
format_workspace_graph_mermaid,
)
ws = analyze_workspace(project_path)
graph = build_workspace_dep_graph(ws)
packages = ws.packages if hasattr(ws, "packages") else []
nodes = [pkg.name if hasattr(pkg, "name") else str(pkg) for pkg in packages]
ws_name = project_path.name
mermaid_str = None
if format == "mermaid":
mermaid_str = format_workspace_graph_mermaid(ws)
text = self.render_ws_text(
ws_name=ws_name, graph=graph, mermaid_str=mermaid_str
)
data: dict[str, object] = {"graph": graph, "nodes": nodes}
if format == "mermaid":
data["mermaid"] = mermaid_str
elif format == "text":
data["text"] = self._format_text(nodes, graph)
return ToolResult(success=True, data=data, text=text)
def _execute_package(self, project_path: Path, *, format: str) -> ToolResult:
"""Build intra-package import graph for a single package."""
from axm_ast.core.analyzer import build_import_graph, module_dotted_name
from axm_ast.core.cache import get_package
from axm_ast.formatters import format_mermaid
pkg = get_package(project_path)
graph = build_import_graph(pkg)
nodes = [module_dotted_name(mod.path, pkg.root) for mod in pkg.modules]
data: dict[str, object] = {"graph": graph, "nodes": nodes}
mermaid_str = None
if format == "mermaid":
mermaid_str = format_mermaid(pkg)
data["mermaid"] = mermaid_str
elif format == "text":
data["text"] = self._format_text(nodes, graph)
pkg_name = getattr(pkg, "name", project_path.name)
text = self.render_pkg_text(
pkg_name=pkg_name,
nodes=nodes,
graph=graph,
mermaid_str=mermaid_str,
)
return ToolResult(success=True, data=data, text=text)
@staticmethod
def _group_modules(
nodes: list[str],
) -> tuple[list[str], dict[str, list[str]]]:
"""Split node names into standalone and dot-grouped buckets."""
groups: dict[str, list[str]] = {}
standalone: list[str] = []
for name in nodes:
parts = name.split(".", 1)
if "." in name:
groups.setdefault(parts[0], []).append(parts[1])
else:
standalone.append(name)
return standalone, groups
@staticmethod
def render_pkg_text(
pkg_name: str,
nodes: list[str],
graph: dict[str, list[str]],
mermaid_str: str | None,
) -> str:
"""Render package graph as compact text for LLM consumption."""
edge_count = sum(len(v) for v in graph.values())
mod_label = "module" if len(nodes) == 1 else "modules"
header = (
f"ast_graph | {pkg_name}"
f" | {len(nodes)} {mod_label} \u00b7 {edge_count} edges"
)
lines = [header]
standalone, groups = GraphTool._group_modules(nodes)
lines.append("Modules:")
for name in standalone:
lines.append(f" {name}")
for prefix, children in groups.items():
lines.append(f" {prefix}: {' '.join(children)}")
# Edges section — omit when empty
if graph:
lines.append("Edges:")
for src, targets in graph.items():
lines.append(f" {src} \u2192 {', '.join(targets)}")
# Mermaid appendix — suppress when zero edges
if mermaid_str and graph:
lines.append("")
lines.append("```mermaid")
lines.append(mermaid_str)
lines.append("```")
return "\n".join(lines)
@staticmethod
def render_ws_text(
ws_name: str,
graph: dict[str, list[str]],
mermaid_str: str | None,
) -> str:
"""Render workspace graph as compact text for LLM consumption."""
all_pkgs: set[str] = set()
for src, targets in graph.items():
all_pkgs.add(src)
all_pkgs.update(targets)
edge_count = sum(len(v) for v in graph.values())
header = (
f"ast_graph | {ws_name} workspace"
f" | {len(all_pkgs)} packages \u00b7 {edge_count} edges"
)
lines = [header]
if graph:
lines.append("Dependencies:")
for src, targets in graph.items():
lines.append(f" {src} \u2192 {', '.join(targets)}")
if mermaid_str:
lines.append("")
lines.append("```mermaid")
lines.append(mermaid_str)
lines.append("```")
return "\n".join(lines)
@staticmethod
def _format_text(nodes: list[str], graph: dict[str, list[str]]) -> str:
"""Render nodes and edges as a human-readable text block."""
lines = ["Nodes:"]
for name in nodes:
lines.append(f" {name}")
lines.append("")
lines.append("Edges:")
for src, targets in graph.items():
for target in targets:
lines.append(f" {src} -> {target}")
return "\n".join(lines)