Skip to content

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.

ConfiguredReferenceFetcher 

ConfiguredReferenceFetcher(
    *, base_url=None, authorization=None
)

Fetcher built from attachments.base_url / attachments.authorization.

  • Relative URLs (/attachments/…) resolve against base_url.
  • The authorization token is attached only to URLs under base_url; any other URL is fetched with the unauthenticated default, so host credentials never leak to third-party origins.

authorization spec formats:

  • forwarded_props:<key> — read the token from the current run's forwarded props (a run-scoped credential the host sends per invocation).
  • env:<VAR> — read a static token from the environment.
Source code in src/kaboo_workflows/tools/fetching.py 
132
133
134
def __init__(self, *, base_url: str | None = None, authorization: str | None = None) -> None:
    self._base_url = base_url.rstrip("/") if base_url else None
    self._authorization = authorization

ReferenceFetcher

Bases: Protocol

Strategy for resolving a reference URL to raw bytes.

Implementations return None on any failure (the caller treats the reference as unresolvable). reference is provided when the fetch is on behalf of a parsed reference; URL-only call sites (entry-message media parts) pass None.

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 
 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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
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

fetch_reference_bytes 

fetch_reference_bytes(url, *, reference=None)

Fetch reference bytes via the registered fetcher (default when none).

Source code in src/kaboo_workflows/tools/fetching.py 
88
89
90
91
92
def fetch_reference_bytes(url: str, *, reference: Reference | None = None) -> bytes | None:
    """Fetch reference bytes via the registered fetcher (default when none)."""
    if _fetcher is not None:
        return _fetcher(url, reference=reference)
    return default_fetch(url)

get_reference_fetcher 

get_reference_fetcher()

Return the registered reference fetcher (None when default).

Source code in src/kaboo_workflows/tools/fetching.py 
83
84
85
def get_reference_fetcher() -> ReferenceFetcher | None:
    """Return the registered reference fetcher (``None`` when default)."""
    return _fetcher

set_reference_fetcher 

set_reference_fetcher(fetcher)

Register the process-wide reference fetcher (None restores default).

Call once at startup, before the app serves requests. Per-run behavior belongs inside the fetcher (read the request context there), not in re-registration.

Source code in src/kaboo_workflows/tools/fetching.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
def set_reference_fetcher(fetcher: ReferenceFetcher | None) -> None:
    """Register the process-wide reference fetcher (``None`` restores default).

    Call once at startup, before the app serves requests. Per-run behavior
    belongs *inside* the fetcher (read the request context there), not in
    re-registration.
    """
    global _fetcher
    if fetcher is not None and not callable(fetcher):
        raise TypeError(
            f"reference fetcher must be callable or None, got {type(fetcher).__name__}."
        )
    _fetcher = fetcher

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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
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 
247
248
249
250
251
252
253
254
255
256
257
258
259
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

fetch_attachment 

fetch_attachment(reference_id)

Fetch a file attachment's contents by its id.

Text-like files (txt, md, csv, html, json, xml) are returned as text. Images, PDFs/office docs, and video are placed into the conversation as media the model can see directly — the tool confirms in text and the file appears in context on the next turn. Non-file references (custom entities) are not fetchable here — resolve those with the tool provided for their kind.

Parameters:

Name Type Description Default
reference_id str

The reference id from the manifest / list_references.

 required
Source code in src/kaboo_workflows/tools/references.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
@tool(name="fetch_attachment")
def fetch_attachment(reference_id: str) -> dict:
    """Fetch a file attachment's contents by its id.

    Text-like files (txt, md, csv, html, json, xml) are returned as text.
    Images, PDFs/office docs, and video are placed into the conversation as
    media the model can see directly — the tool confirms in text and the file
    appears in context on the next turn. Non-file references (custom entities)
    are not fetchable here — resolve those with the tool provided for their kind.

    Args:
        reference_id: The reference id from the manifest / ``list_references``.
    """
    ref = _find(reference_id)
    if ref is None:
        return {
            "status": "error",
            "content": [{"text": f"No reference with id {reference_id!r} was provided."}],
        }
    if ref.transport != "attachment":
        return {
            "status": "error",
            "content": [
                {
                    "text": (
                        f"Reference {ref.id!r} is a '{ref.kind}' item, not a file. "
                        "Resolve it with the tool provided for that kind."
                    )
                }
            ],
        }

    raw = _resolve_bytes(ref)
    if raw is None:
        return {
            "status": "error",
            "content": [{"text": f"Could not resolve the bytes for {ref.id!r}."}],
        }

    if _is_text_mime(ref.mime_type):
        return {"status": "success", "content": [{"text": raw.decode("utf-8", "replace")}]}

    # Media (image/document/video) can't be returned in a tool-role message on
    # OpenAI-compatible providers, so route it through the same user-message
    # injection the ``inline`` policy uses (works across providers). The tool
    # itself returns only text — the file lands in context before the next turn.
    if _content_block(ref, raw) is not None:
        request_inline(ref.id)
        return {
            "status": "success",
            "content": [
                {
                    "text": (
                        f"Loaded '{ref.name}' ({ref.mime_type or 'file'}). It is now "
                        "included in the conversation — read it directly to answer."
                    )
                }
            ],
        }

    return {
        "status": "error",
        "content": [
            {
                "text": (
                    f"Reference {ref.id!r} ({ref.mime_type or 'unknown type'}) is not a "
                    "supported inline media type and has no text representation."
                )
            }
        ],
    }

list_references 

list_references()

List the files and items the user attached or referenced this turn.

Returns each reference's kind, name, id, and transport so you can pick which one to fetch or query. Use the id with fetch_attachment (files) or the kind's dedicated tool (custom items).

Source code in src/kaboo_workflows/tools/references.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@tool(name="list_references")
def list_references() -> dict:
    """List the files and items the user attached or referenced this turn.

    Returns each reference's kind, name, id, and transport so you can pick which
    one to fetch or query. Use the id with ``fetch_attachment`` (files) or the
    kind's dedicated tool (custom items).
    """
    references = get_references()
    if not references:
        return {"status": "success", "content": [{"text": "No references were provided."}]}
    lines = [
        f"- [{r.kind}] {r.name} (id={r.id}, transport={r.transport}"
        + (f", type={r.mime_type}" if r.mime_type else "")
        + ")"
        for r in references
    ]
    return {"status": "success", "content": [{"text": "\n".join(lines)}]}

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 
151
152
153
154
155
156
157
158
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 
141
142
143
144
145
146
147
148
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)