Skip to content

Flows

flows

FlowsTool — execution flow tracing with entry point detection.

FlowsTool

Bases: AXMTool

Trace execution flows and detect entry points in a Python package.

Registered as ast_flows via axm.tools entry point.

Source code in packages/axm-ast/src/axm_ast/tools/flows.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class FlowsTool(AXMTool):
    """Trace execution flows and detect entry points in a Python package.

    Registered as ``ast_flows`` via axm.tools entry point.
    """

    @property
    def name(self) -> str:
        """Return tool name for registry lookup."""
        return "ast_flows"

    def execute(  # noqa: PLR0913
        self,
        *,
        path: str = ".",
        entry: str | None = None,
        max_depth: int = 5,
        cross_module: bool = False,
        detail: str = "trace",
        exclude_stdlib: bool = True,
        **kwargs: Any,
    ) -> ToolResult:
        """Detect entry points or trace flows from a symbol.

        Without ``entry``: returns detected entry points.
        With ``entry``: traces BFS flow from that entry point.

        Args:
            path: Path to package directory.
            entry: Optional entry point name to trace from.
            max_depth: Maximum BFS depth for flow tracing.
            cross_module: Resolve imports and trace into external modules.
            detail: Level of detail — ``"trace"`` (default) or
                ``"source"`` (includes function source code).
            exclude_stdlib: If False, include stdlib/builtin callees
                in the BFS trace.  Default True (exclude them).

        Returns:
            ToolResult with entry points or flow steps.
        """
        try:
            pkg_path = Path(path).resolve()
            if not pkg_path.is_dir():
                return ToolResult(success=False, error=f"Not a directory: {pkg_path}")

            from axm_ast.core.cache import get_package
            from axm_ast.core.flows import find_entry_points, trace_flow

            pkg = get_package(pkg_path)

            if entry is not None:
                steps = trace_flow(
                    pkg,
                    entry,
                    max_depth=max_depth,
                    cross_module=cross_module,
                    detail=detail,
                    exclude_stdlib=exclude_stdlib,
                )
                step_dicts = []
                for s in steps:
                    d: dict[str, object] = {
                        "name": s.name,
                        "module": s.module,
                        "line": s.line,
                        "depth": s.depth,
                        "chain": s.chain,
                    }
                    if s.resolved_module is not None:
                        d["resolved_module"] = s.resolved_module
                    if s.source is not None:
                        d["source"] = s.source
                    step_dicts.append(d)
                return ToolResult(
                    success=True,
                    data={
                        "entry": entry,
                        "steps": step_dicts,
                        "depth": max_depth,
                        "cross_module": cross_module,
                        "count": len(steps),
                    },
                    hint="Tip: Use ast_impact(symbol) on key nodes for blast radius.",
                )

            entries = find_entry_points(pkg)
            return ToolResult(
                success=True,
                data={
                    "entry_points": [
                        {
                            "name": e.name,
                            "module": e.module,
                            "kind": e.kind,
                            "line": e.line,
                            "framework": e.framework,
                        }
                        for e in entries
                    ],
                    "count": len(entries),
                },
                hint="Tip: Use ast_flows(entry='<name>') to trace a flow.",
            )
        except Exception as exc:
            return ToolResult(success=False, error=str(exc))
name property

Return tool name for registry lookup.

execute(*, path='.', entry=None, max_depth=5, cross_module=False, detail='trace', exclude_stdlib=True, **kwargs)

Detect entry points or trace flows from a symbol.

Without entry: returns detected entry points. With entry: traces BFS flow from that entry point.

Parameters:

Name Type Description Default
path str

Path to package directory.

 '.'
entry str | None

Optional entry point name to trace from.

 None
max_depth int

Maximum BFS depth for flow tracing.

 5
cross_module bool

Resolve imports and trace into external modules.

 False
detail str

Level of detail — "trace" (default) or "source" (includes function source code).

 'trace'
exclude_stdlib bool

If False, include stdlib/builtin callees in the BFS trace. Default True (exclude them).

 True

Returns:

Type Description
ToolResult

ToolResult with entry points or flow steps.
Source code in packages/axm-ast/src/axm_ast/tools/flows.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def execute(  # noqa: PLR0913
    self,
    *,
    path: str = ".",
    entry: str | None = None,
    max_depth: int = 5,
    cross_module: bool = False,
    detail: str = "trace",
    exclude_stdlib: bool = True,
    **kwargs: Any,
) -> ToolResult:
    """Detect entry points or trace flows from a symbol.

    Without ``entry``: returns detected entry points.
    With ``entry``: traces BFS flow from that entry point.

    Args:
        path: Path to package directory.
        entry: Optional entry point name to trace from.
        max_depth: Maximum BFS depth for flow tracing.
        cross_module: Resolve imports and trace into external modules.
        detail: Level of detail — ``"trace"`` (default) or
            ``"source"`` (includes function source code).
        exclude_stdlib: If False, include stdlib/builtin callees
            in the BFS trace.  Default True (exclude them).

    Returns:
        ToolResult with entry points or flow steps.
    """
    try:
        pkg_path = Path(path).resolve()
        if not pkg_path.is_dir():
            return ToolResult(success=False, error=f"Not a directory: {pkg_path}")

        from axm_ast.core.cache import get_package
        from axm_ast.core.flows import find_entry_points, trace_flow

        pkg = get_package(pkg_path)

        if entry is not None:
            steps = trace_flow(
                pkg,
                entry,
                max_depth=max_depth,
                cross_module=cross_module,
                detail=detail,
                exclude_stdlib=exclude_stdlib,
            )
            step_dicts = []
            for s in steps:
                d: dict[str, object] = {
                    "name": s.name,
                    "module": s.module,
                    "line": s.line,
                    "depth": s.depth,
                    "chain": s.chain,
                }
                if s.resolved_module is not None:
                    d["resolved_module"] = s.resolved_module
                if s.source is not None:
                    d["source"] = s.source
                step_dicts.append(d)
            return ToolResult(
                success=True,
                data={
                    "entry": entry,
                    "steps": step_dicts,
                    "depth": max_depth,
                    "cross_module": cross_module,
                    "count": len(steps),
                },
                hint="Tip: Use ast_impact(symbol) on key nodes for blast radius.",
            )

        entries = find_entry_points(pkg)
        return ToolResult(
            success=True,
            data={
                "entry_points": [
                    {
                        "name": e.name,
                        "module": e.module,
                        "kind": e.kind,
                        "line": e.line,
                        "framework": e.framework,
                    }
                    for e in entries
                ],
                "count": len(entries),
            },
            hint="Tip: Use ast_flows(entry='<name>') to trace a flow.",
        )
    except Exception as exc:
        return ToolResult(success=False, error=str(exc))