nemo_flow.tools#

Tool lifecycle helpers.

Use this module when you want NeMo Flow to emit tool start and end events around a piece of application logic.

execute() is the usual entry point and runs the full middleware pipeline. call() and call_end() are the lower-level manual lifecycle APIs.

Example:

import nemo_flow

async def search(args):
    return {"result": args["query"].upper()}

result = await nemo_flow.tools.execute("search", {"query": "hello"}, search)
assert result == {"result": "HELLO"}

Functions#

`call`(name, args, *[, handle, attributes, data, ...])	Start a manual tool span and return its `ToolHandle`.
`call_end`(handle, result, *[, data, metadata, timestamp])	Finish a manual tool span started by `call()`.
`execute`(name, args, func, *[, handle, attributes, ...])	Run a tool through the managed middleware pipeline.
`request_intercepts`(name, args)	Apply global tool request intercepts to `args`.
`conditional_execution`(name, args)	Run tool conditional-execution guardrails for `args`.

Module Contents#

nemo_flow.tools.call( name, args, *, handle=None, attributes=None, data=None, metadata=None, tool_call_id=None, timestamp: datetime.datetime | None = None, )#

Start a manual tool span and return its ToolHandle.

Parameters:

name – Tool name recorded on emitted lifecycle events.
args – JSON-compatible tool arguments to associate with the call.
handle – Optional parent scope handle. When omitted, the current scope becomes the parent.
attributes – Optional native tool attributes attached to the start event.
data – Optional JSON application payload stored on the tool handle.
metadata – Optional JSON metadata recorded on the emitted start event.
tool_call_id – Optional provider-specific tool call identifier to attach to the emitted events.
timestamp – Optional timezone-aware datetime recorded as the handle start time and on the emitted start event. When omitted, the current runtime time is used.

Returns:

Handle used to finish the manual span with call_end().

Return type:

ToolHandle

Notes

This starts only the manual tool lifecycle span. It applies sanitize-request guardrails to the emitted start-event payload but does not run request or execution intercepts. timestamp must be a timezone-aware datetime; strings and naive datetimes are rejected.

Example:

import nemo_flow

handle = nemo_flow.tools.call(
    "search",
    {"query": "hello"},
    handle=None,
    attributes=None,
    data={"attempt": 1},
    metadata={"path": "manual"},
    tool_call_id="tool-call-1",
)
nemo_flow.tools.call_end(
    handle,
    {"result": "ok"},
    data={"cached": False},
    metadata={"status": "success"},
)

nemo_flow.tools.call_end( handle, result, *, data=None, metadata=None, timestamp: datetime.datetime | None = None, )#

Finish a manual tool span started by call().

Parameters:

handle – Tool handle returned by call().
result – JSON-compatible tool result to record on the end event.
data – Optional JSON payload used when the sanitized result is JSON null.
metadata – Optional JSON metadata recorded on the emitted end event.
timestamp – Optional timezone-aware datetime recorded on the emitted end event. When omitted, the runtime default end timestamp is used.

Returns:

This function returns after the end event has been recorded.

Return type:

None

Notes

call_end() applies sanitize-response guardrails to the emitted end-event payload but does not alter the caller-owned result object. timestamp must be a timezone-aware datetime; strings and naive datetimes are rejected.

nemo_flow.tools.execute( name, args, func, *, handle=None, attributes=None, data=None, metadata=None, )#

Run a tool through the managed middleware pipeline.

Pipeline order:

tool conditional-execution guardrails
tool request intercepts
tool sanitize-request guardrails for emitted start events
tool execution intercepts
func(args)
tool sanitize-response guardrails for emitted end events

Parameters:

name – Tool name recorded on emitted lifecycle events.
args – JSON-compatible arguments passed through the middleware pipeline.
func – Tool implementation invoked as func(args) after guardrails and intercepts run.
handle – Optional parent scope handle. When omitted, the current scope becomes the parent.
attributes – Optional native tool attributes attached to the start event.
data – Optional JSON application payload stored on the managed tool handle.
metadata – Optional JSON metadata recorded on the emitted start event.

Returns:

The raw result returned by func or by an execution intercept.

Return type:

Json

Notes

Sanitize guardrails affect emitted event payloads only. They do not mutate the arguments passed to func or the value returned to the caller.

Example:

import nemo_flow

async def local_tool(args):
    return {"count": len(args["items"])}

result = await nemo_flow.tools.execute(
    "count",
    {"items": [1, 2, 3]},
    local_tool,
    handle=None,
    attributes=None,
    data={"source": "example"},
    metadata={"request_id": "req-1"},
)
assert result["count"] == 3

nemo_flow.tools.request_intercepts(name, args)#

Apply global tool request intercepts to args.

Parameters:

name – Tool name used when evaluating the registered intercept chain.
args – JSON-compatible tool arguments to pass through the intercepts.

Returns:

The arguments produced by the final request intercept.

Return type:

Json

Notes

This runs only the request-intercept chain. It does not execute conditional guardrails, sanitize guardrails, or the tool callback.

nemo_flow.tools.conditional_execution(name, args)#

Run tool conditional-execution guardrails for args.

Parameters:

name – Tool name used when evaluating registered guardrails.
args – JSON-compatible tool arguments to validate.

Returns:

A rejection message if execution should be blocked, otherwise None.

Return type:

str | None

Notes

This helper evaluates only the conditional-execution guardrail chain and does not invoke request intercepts or tool execution.