Tools

A tool is a regular Python callable the agent can invoke. The @tool decorator derives its JSON schema from type hints + docstring. Async callables are awaited: async def functions, a functools.partial of one, and objects with an async def __call__. Sync functions are dispatched to a worker thread so they never block the event loop; when a sync call times out or is cancelled, the run moves on at once and the thread finishes in the background.

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: its whole body, minus Args: / Returns: / Raises: (and similar) sections, not just the first line. Entries of an Args: (Google style) or Parameters (NumPy style) section become the matching parameters’ descriptions in the schema. An explicit @tool(description=...) is used exactly as written.

@tool
async def search(query: str, limit: int = 5) -> list[str]:
    """Search the knowledge base.
 
    Results are ranked by relevance; prefer specific queries.
 
    Args:
        query: What to look for.
        limit: How many results to return.
    """

The tool’s description is both paragraphs (“Search the knowledge base.” and “Results are ranked by relevance; …”), and the schema gives query and limit their descriptions from the Args: entries.

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), a destructive call gets PermissionDecision.ask_(...), which is routed to the agent’s approval_handler.
• With Mode.ACCEPT_EDITS, only the file-edit tools auto-approve (the built-in write and edit, or whatever you list in StandardPermissions(edit_tools=[...])); every other destructive tool still asks.
• With Mode.BYPASS, every call passes (CI / sandbox only).

See Approval handlers.

Tools that end the turn

Some tools hand the work to someone else, after which the agent has nothing more to say. Mark them ends_turn=True and a successful call ends the agent’s run right there instead of calling the model again:

from loomflow import Agent, tool
 
@tool(ends_turn=True)
async def escalate_to_human(reason: str) -> str:
    """Hand this conversation to a human agent."""
    return "escalated"   # open the ticket here
 
agent = Agent("You are first-line support.", model="claude-opus-4-7", tools=[escalate_to_human])

The run’s output is whatever the model said alongside the call, and stop hooks and output_schema validation are skipped for that run. If the call fails, the agent carries on as usual. Swarm handoffs work this way.

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. It has three methods: list_tools, call and watch.

Where to next