Skip to content

Node

node

Adapt an :class:~axm.tools.base.AXMTool into a DAG python-node function.

This is the third consumer of the single axm.tools declaration: the same tool that MCP exposes and the CLI auto-generates can be called directly from a DAG node, with no HookAction and no subprocess. One entry point → MCP + CLI + node (see synchronisation_cmp_cli/README.md, révision axm-dag).

A DAG python node is a callable fn(payload) -> dict whose returned keys are the node's writes. :func:tool_node builds such a callable around a tool:

  • inputs — the node's reads arrive in payload; they map to the tool's execute(**kwargs) by name, or via an explicit args rename map when the mem key differs from the parameter name;
  • outputs — returns maps each write key to its source: the literal "text" (the tool's ToolResult.text) or a key inside ToolResult.data;
  • failure — fail-fast: a tool returning success=False raises :class:ToolNodeError. Guard preconditions with a conditional node (router / if_) so the tool is only invoked when it can succeed.

ToolNodeError

Bases: RuntimeError

A tool invoked as a DAG node failed (ToolResult.success was False).

Source code in packages/axm/src/axm/tools/node.py
Python
40
41
class ToolNodeError(RuntimeError):
    """A tool invoked as a DAG node failed (``ToolResult.success`` was ``False``)."""

tool_node(name, *, args=None, returns=None)

Build a DAG python-node fn(payload) -> dict around an axm.tools tool.

Parameters:

Name Type Description Default
name str

The tool's axm.tools entry-point name (e.g. "ast_impact").

 required
args Mapping[str, str] | None

Optional {param_name: mem_key} rename map for params whose mem key differs from the execute parameter name. Params absent from args are read from payload by their own name when present.

 None
returns Mapping[str, str] | None

{write_key: source} where source is "text" (the ToolResult.text) or a key inside ToolResult.data. Defaults to {name: "text"} is not assumed — supply it so the node's writes are explicit (decision: writes nommés par clé).

 None

Returns:

Type Description
Callable[[Mapping[str, object]], dict[str, object]]

A callable mapping the node's payload to a dict keyed by returns.

Raises:

Type Description
ToolNodeError

At call time, if the tool is unknown or returns success=False (fail-fast).
Source code in packages/axm/src/axm/tools/node.py
Python
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
def tool_node(
    name: str,
    *,
    args: Mapping[str, str] | None = None,
    returns: Mapping[str, str] | None = None,
) -> Callable[[Mapping[str, object]], dict[str, object]]:
    """Build a DAG python-node ``fn(payload) -> dict`` around an ``axm.tools`` tool.

    Args:
        name: The tool's ``axm.tools`` entry-point name (e.g. ``"ast_impact"``).
        args: Optional ``{param_name: mem_key}`` rename map for params whose mem
            key differs from the ``execute`` parameter name. Params absent from
            *args* are read from ``payload`` by their own name when present.
        returns: ``{write_key: source}`` where *source* is ``"text"`` (the
            ``ToolResult.text``) or a key inside ``ToolResult.data``. Defaults to
            ``{name: "text"}`` is **not** assumed — supply it so the node's writes
            are explicit (decision: writes nommés par clé).

    Returns:
        A callable mapping the node's ``payload`` to a dict keyed by *returns*.

    Raises:
        ToolNodeError: At call time, if the tool is unknown or returns
            ``success=False`` (fail-fast).
    """
    rename = dict(args or {})
    out_spec = dict(returns or {})

    def _run(payload: Mapping[str, object]) -> dict[str, object]:
        tool = _load_tool(name)
        kwargs = _kwargs_from_payload(payload, rename)
        result = tool.execute(**kwargs)
        if not result.success:
            msg = f"tool {name!r} failed: {result.error or '<no error message>'}"
            raise ToolNodeError(msg)
        return _shape_output(name, result.data, result.text, out_spec)

    return _run