Tools & Retrievers

Tools let a ReActAgent call your Python functions during its reasoning loop. On each iteration the LLM decides whether to answer directly or to invoke one of the tools you registered. When it chooses a tool, the framework runs the function, captures the return value, and feeds it back to the LLM as an observation — the ReAct pattern: call → observe → iterate.

Concretely, every loop iteration looks like this:

1. The LLM streams a response. If it wants a tool, it emits a ToolCall (the tool name plus JSON-encoded arguments).
2. The framework looks the name up in the agent's tool registry and runs the function with the decoded arguments.
3. The return value becomes a ToolResult, persisted into history as a Message(role="tool", tool_call_id=...) and shown back to the LLM.
4. The loop repeats until the LLM produces final text (no tool calls) or max_iterations is reached.

from latent.agents import ReActAgent, tool
from latent.protocol import Message

Async-only since 5.0.0

Agents are invoked via await agent.invoke(...), agent.stream(...), or await agent.ask(...). Tools themselves must be async def (see Sync vs async).

---

@tool on a function

Decorate an async def with @tool to expose it to the LLM. Two things are required:

• A docstring. It becomes the tool's description — the text the LLM reads to decide when to call it. A function with no docstring raises ValueError at decoration time.
• Type-annotated parameters. Each parameter's annotation is mapped to a JSON Schema type to build the tool's parameters schema.

Python types map to JSON Schema as follows; any unrecognized annotation falls back to "string":

Python type	JSON Schema type
str	string
int	integer
float	number
bool	boolean
list	array
dict	object

Parameters without a default are added to the schema's required list; parameters with a default are optional. A self parameter is skipped (relevant for method tools).

from latent.agents import ReActAgent, tool


@tool
async def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"It is 22C and sunny in {city}."


@tool
async def convert_currency(amount: float, to_currency: str) -> str:
    """Convert a USD amount into another currency."""
    rates = {"EUR": 0.92, "GBP": 0.79}
    return f"{amount * rates[to_currency]:.2f} {to_currency}"

Pass the decorated functions to the agent via the tools=[...] constructor argument:

agent = ReActAgent(
    name="assistant",
    model="gpt-4o",
    system_prompt="You are a helpful assistant. Use tools when relevant.",
    tools=[get_weather, convert_currency],
)

answer = await agent.ask("What's the weather in Paris, and convert 50 USD to EUR?")

The agent inspects each passed function for the _tool_meta stamped by @tool. A function missing _tool_meta (i.e. not decorated with @tool or @retriever) raises ValueError at construction.

You can inspect the resulting schemas via the read-only agent.tools property, which returns the OpenAI-style function definitions sent to the model.

---

@tool methods on a subclass

Tools can also live as methods on a ReActAgent subclass. Any @tool-decorated method is auto-discovered at construction — you do not need to list it in tools=. The self parameter is ignored when building the schema.

from latent.agents import ReActAgent, tool


class SupportAgent(ReActAgent):
    @tool
    async def lookup_order(self, order_id: str) -> str:
        """Look up the status of a customer order by its ID."""
        return await self._db.fetch_order(order_id)

    @tool
    async def issue_refund(self, order_id: str, amount: float) -> str:
        """Issue a refund for an order."""
        return await self._db.refund(order_id, amount)


agent = SupportAgent(
    name="support",
    model="gpt-4o",
    system_prompt="You help customers with orders and refunds.",
)
# lookup_order and issue_refund are already registered.

You can mix both styles. Tools passed via tools=[...] override an auto-discovered method tool on a name clash (the explicitly-passed function wins).

---

Sync vs async tools

Tools must be async def. Sync tools were dropped in 5.4.0 — the agent pipeline is async end-to-end. A plain (non-coroutine, non-async-generator) function raises ValueError when decorated:

@tool
def broken(x: str) -> str:   # raises ValueError: must be ``async def``
    """..."""
    return x

Internally, @tool unifies two shapes so the loop can treat every tool the same way:

• A coroutine tool (async def f(...): return value) is wrapped to yield its return value once.
• An async-generator tool (one that yields — used when a tool forwards retriever events; see below) passes through unchanged.

The ReAct loop drains every tool with async for, awaiting/iterating it automatically. You write ordinary async def tools and await whatever you need inside them; the framework handles the rest.

---

The end_call tool / END_CALL_TOOL_NAME

END_CALL_TOOL_NAME is a constant ("end_call") naming the tool an agent calls to signal that a conversation has reached its natural end. It is a convention, not something automatically registered on every ReActAgent.

from latent.agents import END_CALL_TOOL_NAME  # == "end_call"

The shipped end_call tool lives in the human-simulation package and is a normal @tool:

@tool
async def end_call(reason: str) -> str:
    """End the current conversation. Call this when the conversation has reached a natural conclusion."""
    return f"Call ended: {reason}"

HumanAgent registers it (in goal-directed and free-form modes) so the simulated human can choose to hang up. In conversation simulation, the helper respond_from_events(agent, messages) runs the agent, collects its tool calls, and sets should_end=True on the returned AgentResponse when any tool call matches end_tool_name (defaulting to END_CALL_TOOL_NAME). To use this pattern in your own agent, register a tool named "end_call" (or pass a custom end_tool_name) and check AgentResponse.should_end to terminate the loop.

---

@retriever

@retriever decorates an async def retrieval function so it emits RetrieverStart / RetrieverEnd events around the underlying call — turning an opaque vector-store / BM25 lookup into observable activity in the event stream.

The contract:

• The function must be an async def (not itself an async generator — that raises TypeError).
• It should take an argument named query (the search input). Other names like action_query or queries are auto-detected as the query parameter, and the first non-self parameter is used as a fallback. This name only affects the event payload, not the tool schema.
• It should return an iterable of objects with a content attribute (e.g. retrieved chunks) or items that are str-convertible; a raw str counts as a single document.

from latent.agents import retriever


@retriever
async def search_kb(query: str, k: int = 4):
    """Search the knowledge base for relevant chunks."""
    return await chroma.retrieve(query, k=k)

When called, the decorated function is an async generator: it yields a RetrieverStart (carrying run_id, query, and name), then awaits the underlying retrieval, then yields a RetrieverEnd (carrying the same run_id, a name, and documents — a list[str] content preview), and finally yields the actual retrieval result as the last non-event item. If the retrieval raises, a RetrieverEnd with empty documents is emitted before the exception propagates, so every RetrieverStart is paired.

By default the retriever name comes from fn.__name__; override the event name with @retriever(name="my-retriever") (the tool name exposed to the LLM still comes from fn.__name__).

@retriever implies @tool

A @retriever-decorated function is also stamped with _tool_meta, so you can pass it straight to ReActAgent(tools=[...]) for the common case where the retriever's return value is the tool output:

agent = ReActAgent(
    name="rag",
    model="gpt-4o",
    system_prompt="Answer using the knowledge base.",
    tools=[search_kb],   # @retriever is also a tool — no separate @tool needed
)

Composing retrievers inside a tool

For the case where one tool fans out to multiple retrievers and post-processes the results, write a separate @tool async generator that drains the @retriever building blocks. Forward events with yield, and yield the final tool output last:

from latent.agents import tool, retriever
from latent.events import AgentEvent


@retriever
async def search_kb(query: str, k: int = 4):
    """Search the knowledge base."""
    return await chroma.retrieve(query, k=k)


@tool
async def answer_from_kb(query: str):
    """Retrieve supporting passages and format them for the model."""
    chunks = None
    async for item in search_kb(query):
        if isinstance(item, AgentEvent):
            yield item          # forward RetrieverStart / RetrieverEnd
        else:
            chunks = item       # last non-event yield = retriever result
    yield format_chunks(chunks)  # tool's final yield = the ToolResult output

An async-generator tool must yield exactly one non-event item as its result. Yielding only events raises a RuntimeError ("produced no result yield").

---

How tool activity appears in the event stream

Tool and retriever activity surfaces as typed events while you stream() the agent (see the Agent Protocol for the full event taxonomy):

Event	Meaning
ToolCall	The LLM is invoking a tool. Has id, type="function", and a function block (FunctionCall) with function.name and function.arguments (a JSON-encoded string, not a dict).
ToolResult	The outcome. Carries tool_call_id (matching the ToolCall.id), output (Python-native), and error (str | None).
RetrieverStart	A retriever began a query: run_id, query, name.
RetrieverEnd	A retriever finished: matching run_id, documents (preview), name.

Parse a tool call's arguments with json.loads (or use ToolCall.from_arguments(...) to build one from a dict). When a step is persisted into history, the assistant's calls become Message(role="assistant", tool_calls=[ToolCall, ...]) and each result becomes a separate Message(role="tool", tool_call_id=..., content=str(output)) row.

import json
from latent.protocol import ToolCall, ToolResult

async for event in agent.stream([Message(role="user", content="weather in Tokyo?")]):
    if isinstance(event, ToolCall):
        args = json.loads(event.function.arguments or "{}")
        print("calling", event.function.name, "with", args)
    elif isinstance(event, ToolResult):
        if event.error:
            print("tool failed:", event.error)
        else:
            print("tool returned:", event.output)

Tool failures don't crash the loop: if argument JSON is invalid, the tool name isn't registered, or the tool raises, the framework emits a ToolResult with error set and feeds an error message back to the LLM so it can recover or re-plan.

---

See Also

• ReAct Agent
• Agent Protocol
• Guardrails