kaboo_workflows.tools

Tool discovery, loading, and node-as-tool adapters.

kaboo_workflows.tools

Tool loading and wrapping utilities.

Provides helpers for: - Loading @tool-decorated functions from files, modules, and directories. - Wrapping Agent / MultiAgentBase nodes as AgentTool instances (node_as_tool, node_as_async_tool) for delegation. - Serializing multi-agent results with full execution metadata.

serialize_multiagent_result

serialize_multiagent_result(result)

Serialize a MultiAgentResult with execution metadata omitted by to_dict().

Extends result.to_dict() with fields only available on the live object:

• last_node_id — id of the truly last executing node, derived from node_history / execution_order (not dict insertion order).
• response — plain-text answer from that node, ready to use directly without any further extraction.
• swarm.node_history — full ordered execution trace including repeated visits (SwarmResult only).
• graph.execution_order, graph.edges, graph.entry_points, and node counts (GraphResult only).

Parameters:

Name	Type	Description	Default
result	MultiAgentResult	A live MultiAgentResult, SwarmResult, or GraphResult returned directly by invoke_async.	required

Returns:

Type	Description
dict[str, Any]	A JSON-serializable dict extending result.to_dict().

Source code in src/kaboo_workflows/tools/extractors.py

def serialize_multiagent_result(result: MultiAgentResult) -> dict[str, Any]:
    """Serialize a ``MultiAgentResult`` with execution metadata omitted by ``to_dict()``.

    Extends ``result.to_dict()`` with fields only available on the live object:

    - ``last_node_id`` — id of the truly last executing node, derived from
      ``node_history`` / ``execution_order`` (not dict insertion order).
    - ``response`` — plain-text answer from that node, ready to use
      directly without any further extraction.
    - ``swarm.node_history`` — full ordered execution trace including repeated
      visits (``SwarmResult`` only).
    - ``graph.execution_order``, ``graph.edges``, ``graph.entry_points``, and
      node counts (``GraphResult`` only).

    Args:
        result: A live ``MultiAgentResult``, ``SwarmResult``, or ``GraphResult``
            returned directly by ``invoke_async``.

    Returns:
        A JSON-serializable dict extending ``result.to_dict()``.
    """
    data = result.to_dict()

    last_node_id = resolve_last_node_id(result)
    data["last_node_id"] = last_node_id

    final_message = extract_last_message(result)
    data["response"] = extract_text(final_message)

    # SwarmResult extras — node_history is a list[SwarmNode]
    node_history: list[Any] | None = getattr(result, "node_history", None)
    if node_history is not None:
        data["swarm"] = {
            "node_history": [str(n.node_id) for n in node_history],
        }

    # GraphResult extras — execution_order, edges, node counts
    execution_order: list[Any] | None = getattr(result, "execution_order", None)
    if execution_order is not None:
        edges_raw: list[Any] = getattr(result, "edges", []) or []
        entry_points_raw: list[Any] = getattr(result, "entry_points", []) or []

        edges: list[list[str]] = []
        for edge in edges_raw:
            if isinstance(edge, tuple) and len(edge) == 2:
                edges.append([str(edge[0].node_id), str(edge[1].node_id)])
            else:
                # GraphEdge dataclass with from_node / to_node attributes
                from_id = str(getattr(getattr(edge, "from_node", None), "node_id", edge))
                to_id = str(getattr(getattr(edge, "to_node", None), "node_id", edge))
                edges.append([from_id, to_id])

        data["graph"] = {
            "execution_order": [str(n.node_id) for n in execution_order],
            "edges": edges,
            "entry_points": [str(getattr(ep, "node_id", ep)) for ep in entry_points_raw],
            "completed_nodes": getattr(result, "completed_nodes", 0),
            "failed_nodes": getattr(result, "failed_nodes", 0),
            "interrupted_nodes": getattr(result, "interrupted_nodes", 0),
        }

    return data

load_tool_function

load_tool_function(spec)

Load a specific tool function from a module.

If the named function is already decorated with @tool it is returned as-is. If it is a plain callable it is automatically wrapped with @tool and a warning is logged — users should consider adding the decorator explicitly.

Parameters:

Name	Type	Description	Default
spec	str	Colon-separated string in "module.path:function_name" format.	required

Returns:

Type	Description
AgentTool	An AgentTool instance for the named function.

Raises:

Type	Description
ValueError	If the spec format is invalid.
ImportError	If the module cannot be imported.
AttributeError	If the function does not exist in the module.
TypeError	If the named attribute is not callable.

Source code in src/kaboo_workflows/tools/loaders.py

def load_tool_function(spec: str) -> AgentTool:
    """Load a specific tool function from a module.

    If the named function is already decorated with ``@tool`` it is returned
    as-is.  If it is a plain callable it is automatically wrapped with
    ``@tool`` and a warning is logged — users should consider adding the
    decorator explicitly.

    Args:
        spec: Colon-separated string in "module.path:function_name" format.

    Returns:
        An ``AgentTool`` instance for the named function.

    Raises:
        ValueError: If the spec format is invalid.
        ImportError: If the module cannot be imported.
        AttributeError: If the function does not exist in the module.
        TypeError: If the named attribute is not callable.
    """
    if ":" not in spec:
        raise ValueError(f"Invalid tool spec format, expected 'module:function': {spec}")

    module_path, function_name = spec.rsplit(":", 1)
    module = importlib.import_module(module_path)
    if not hasattr(module, function_name):
        raise AttributeError(f"Module '{module_path}' has no attribute '{function_name}'")
    return _ensure_tool(getattr(module, function_name), function_name)

load_tools_from_directory

load_tools_from_directory(path)

Load all @tool functions from .py files in a directory.

Scans all .py files (excluding _-prefixed) and loads their tools.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory path to scan.	required

Returns:

Type	Description
list[AgentTool]	List of AgentTool instances from all files in the directory.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.
NotADirectoryError	If the path is not a directory.

Source code in src/kaboo_workflows/tools/loaders.py

def load_tools_from_directory(path: str | Path) -> list[AgentTool]:
    """Load all @tool functions from .py files in a directory.

    Scans all .py files (excluding ``_``-prefixed) and loads their tools.

    Args:
        path: Directory path to scan.

    Returns:
        List of AgentTool instances from all files in the directory.

    Raises:
        FileNotFoundError: If the directory does not exist.
        NotADirectoryError: If the path is not a directory.
    """
    dir_path = Path(path).resolve()
    if not dir_path.exists():
        raise FileNotFoundError(f"Tool directory not found: {dir_path}")
    if not dir_path.is_dir():
        raise NotADirectoryError(f"Path is not a directory: {dir_path}")

    tools: list[AgentTool] = []
    for py_file in sorted(dir_path.glob("*.py")):
        if py_file.name.startswith("_"):
            logger.debug("file=<%s> | skipping underscore-prefixed file", py_file.name)
            continue
        loaded = load_tools_from_file(py_file)
        tools.extend(loaded)
    return tools

load_tools_from_file

load_tools_from_file(path)

Load tools from a Python file.

Imports the file as a module and collects all @tool-decorated objects. Plain functions without the @tool decorator are silently ignored — users must decorate their functions explicitly.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a .py file containing tool functions.	required

Returns:

Type	Description
list[AgentTool]	List of AgentTool instances found in the file.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.
ImportError	If the file cannot be loaded as a module.

Source code in src/kaboo_workflows/tools/loaders.py

def load_tools_from_file(path: str | Path) -> list[AgentTool]:
    """Load tools from a Python file.

    Imports the file as a module and collects all ``@tool``-decorated objects.
    Plain functions without the ``@tool`` decorator are silently ignored —
    users must decorate their functions explicitly.

    Args:
        path: Path to a .py file containing tool functions.

    Returns:
        List of AgentTool instances found in the file.

    Raises:
        FileNotFoundError: If the file does not exist.
        ImportError: If the file cannot be loaded as a module.
    """
    module = load_module_from_file(path)
    return _collect_tools(module)

load_tools_from_module

load_tools_from_module(module_path)

Load tools from a Python module.

First scans for @tool-decorated functions. If none are found, falls back to the strands module-based tool pattern (TOOL_SPEC dict + a function named after the module). This ensures compatibility with tools like strands_tools.http_request that use the legacy pattern.

Parameters:

Name	Type	Description	Default
module_path	str	Dotted import path (e.g., "my_package.tools").	required

Returns:

Type	Description
list[AgentTool]	List of AgentTool instances found in the module.

Raises:

Type	Description
ImportError	If the module cannot be imported.
AttributeError	If the module contains neither @tool-decorated functions nor a valid TOOL_SPEC + module-named function.

Source code in src/kaboo_workflows/tools/loaders.py

def load_tools_from_module(module_path: str) -> list[AgentTool]:
    """Load tools from a Python module.

    First scans for ``@tool``-decorated functions.  If none are found,
    falls back to the strands module-based tool pattern (``TOOL_SPEC`` dict
    + a function named after the module).  This ensures compatibility with
    tools like ``strands_tools.http_request`` that use the legacy pattern.

    Args:
        module_path: Dotted import path (e.g., "my_package.tools").

    Returns:
        List of AgentTool instances found in the module.

    Raises:
        ImportError: If the module cannot be imported.
        AttributeError: If the module contains neither ``@tool``-decorated
            functions nor a valid ``TOOL_SPEC`` + module-named function.
    """
    module = importlib.import_module(module_path)
    tools = _collect_tools(module)
    if tools:
        return tools

    # Fallback: delegate to strands for TOOL_SPEC-based (module) tools.
    # Strands raises AttributeError if the module has neither @tool functions
    # nor a valid TOOL_SPEC + module-named function.
    module_name = module_path.split(".")[-1]
    return _strands_load_tools_from_module(module, module_name)

resolve_tool_spec

resolve_tool_spec(spec)

Resolve a tool specification string to tool objects.

Supported formats (checked in order):

• "./path/to/file.py:function_name" — single tool from a file
• "./path/to/dir/" or "./path/to/dir" (is a directory) — all tools in dir
• "./path/to/file.py" or "path/to/file.py" — all tools from file
• "module.path:function_name" — single tool from module
• "module.path" — all tools from module

The heuristic for path vs. module: if the spec contains /, \\, or ends with .py (before any :) it is treated as a filesystem path.

Parameters:

Name	Type	Description	Default
spec	str	Tool specification string.	required

Returns:

Type	Description
list[AgentTool]	List of tool objects.

Source code in src/kaboo_workflows/tools/loaders.py

def resolve_tool_spec(spec: str) -> list[AgentTool]:
    r"""Resolve a tool specification string to tool objects.

    Supported formats (checked in order):

    - ``"./path/to/file.py:function_name"`` — single tool from a file
    - ``"./path/to/dir/"`` or ``"./path/to/dir"`` (is a directory) — all tools in dir
    - ``"./path/to/file.py"`` or ``"path/to/file.py"`` — all tools from file
    - ``"module.path:function_name"`` — single tool from module
    - ``"module.path"`` — all tools from module

    The heuristic for path vs. module: if the spec contains ``/``, ``\\``, or
    ends with ``.py`` (before any ``:``) it is treated as a filesystem path.

    Args:
        spec: Tool specification string.

    Returns:
        List of tool objects.
    """
    # Determine the path part (before colon, if present).
    # On Windows, absolute paths contain a drive letter colon (e.g. "C:\...").
    # Strip a leading drive prefix so the colon check doesn't misfire.
    _drive = Path(spec).drive  # e.g. "C:" on Windows, "" on POSIX
    _after_drive = spec[len(_drive) :]
    path_part = _after_drive.rsplit(":", 1)[0] if ":" in _after_drive else _after_drive
    is_fs_path = "/" in path_part or "\\" in path_part or path_part.endswith(".py")

    if is_fs_path:
        if ":" in _after_drive:
            # ./path/to/file.py:function_name — single function from a file
            file_str, func_name = _after_drive.rsplit(":", 1)
            file_str = _drive + file_str
            module = load_module_from_file(file_str)
            if not hasattr(module, func_name):
                raise AttributeError(
                    f"Tool file '{Path(file_str).resolve()}' has no attribute '{func_name}'"
                )
            return [_ensure_tool(getattr(module, func_name), func_name)]

        # No colon: file or directory
        candidate = Path(spec)
        if candidate.is_dir() or spec.endswith("/") or spec.endswith("\\"):
            return list(load_tools_from_directory(candidate))

        return list(load_tools_from_file(spec))

    # Module-based specs
    if ":" in spec:
        return [load_tool_function(spec)]

    return list(load_tools_from_module(spec))

resolve_tool_specs

resolve_tool_specs(specs)

Resolve a list of tool specification strings.

Parameters:

Name	Type	Description	Default
specs	list[str]	List of tool specification strings.	required

Returns:

Type	Description
list[AgentTool]	Flat list of all resolved tool objects.

Source code in src/kaboo_workflows/tools/loaders.py

def resolve_tool_specs(specs: list[str]) -> list[AgentTool]:
    """Resolve a list of tool specification strings.

    Args:
        specs: List of tool specification strings.

    Returns:
        Flat list of all resolved tool objects.
    """
    tools: list[AgentTool] = []
    for spec in specs:
        tools.extend(resolve_tool_spec(spec))
    return tools

node_as_async_tool

node_as_async_tool(node, *, name=None, description)

Wrap an Agent or MultiAgentBase as an async AgentTool for delegation.

Source code in src/kaboo_workflows/tools/wrappers.py

def node_as_async_tool(
    node: Node,
    *,
    name: str | None = None,
    description: str,
) -> DecoratedFunctionTool:
    """Wrap an Agent or MultiAgentBase as an async ``AgentTool`` for delegation."""
    return _node_as_tool(node, name=name, description=description, is_async=True)

node_as_tool

node_as_tool(node, *, name=None, description)

Wrap an Agent or MultiAgentBase as a sync AgentTool for delegation.

Source code in src/kaboo_workflows/tools/wrappers.py

def node_as_tool(
    node: Node,
    *,
    name: str | None = None,
    description: str,
) -> DecoratedFunctionTool:
    """Wrap an Agent or MultiAgentBase as a sync ``AgentTool`` for delegation."""
    return _node_as_tool(node, name=name, description=description, is_async=False)