Tools

A tool is a regular Python callable the agent can invoke. The @tool decorator derives its JSON schema from type hints + docstring; sync functions are automatically dispatched to a worker thread so they never block the event loop.

from loomflow import Agent, tool
 
@tool
async def get_weather(city: str) -> str:
    """Look up the current weather for a city."""
    return f"Sunny, 72°F in {city}."
 
@tool(destructive=True)
def delete_file(path: str) -> str:
    """Delete a file. Marked destructive so default permissions ask first."""
    import os
    os.remove(path)
    return f"deleted {path}"
 
agent = Agent(
    "You are a productivity assistant.",
    model="claude-opus-4-7",
    tools=[get_weather, delete_file],
)

Schema derivation

Schemas are fully recursive. A parameter typed as a nested Pydantic model, a list[SomeModel], or any composition of the above resolves through pydantic.TypeAdapter, so deeply nested argument shapes are described to the model exactly. The function’s docstring becomes the tool description.

Arguments the model supplies are coerced to their declared types before the function runs — a numeric value sent as a string, for example, is converted to int or float rather than rejected. The coercion is conservative (only applied when the declared type is unambiguous), so a well-typed tool gets clean Python values without spending turns on validation round-trips.

Parallel dispatch

Tool calls in the same model turn run concurrently under a single anyio.create_task_group. Tool results are appended in arrival order , deterministic for the model’s next turn. A failing tool doesn’t poison the others: each exception is captured in its own ToolResult(ok=False) and the loop continues.

Reading run scope inside a tool

Don’t plumb user_id / session_id through every tool signature. Use get_run_context():

from loomflow import get_run_context, tool
 
@tool
async def fetch_orders() -> str:
    ctx = get_run_context()           # RunContext
    return await db.query("orders", user_id=ctx.user_id)

ctx.user_id and ctx.session_id are always populated for tools called inside an agent run.

Marking tools destructive

@tool(destructive=True) is a hint to the permissions layer:

• With StandardPermissions(mode=Mode.DEFAULT), destructive tools return Decision.ask_(...). Routed through the approval_handler.
• With Mode.ACCEPT_EDITS, destructive tools auto-approve.
• With Mode.BYPASS, every call passes (CI / sandbox only).

See Approval handlers.

Single tool, no list-wrapping

Pass one callable directly:

agent = Agent("...", model="...", tools=get_weather)

Lists still work for multiple tools.

InProcessToolHost

When you need a programmable host (custom tool dispatch logic, dynamic registration, metrics), use InProcessToolHost:

from loomflow import Agent
from loomflow.tools import InProcessToolHost
 
host = InProcessToolHost([get_weather, delete_file])
host.register(extra_tool)
host.unregister("delete_file")
 
agent = Agent("...", model="...", tools=host)

For external tool sources (MCP servers, LangChain bridges) implement the ToolHost protocol directly. Three methods.

Where to next