Adding a real tool

Wire a real backend behind a @tool. Validate its inputs, shape its output for the model, decide its error policy.

The brief writer ships with a stub search tool. This recipe replaces it with a real one. The pattern applies to any @tool — search, lookup, calculation, RPC.

Start from the stub

@tool
def search(query: str) -> list[str]:
    """Search the web. Returns up to 10 result URLs."""
    slug = query.lower().replace(" ", "-")[:40]
    return [f"https://example.com/{slug}/{i}" for i in range(5)]

The model sees one argument (query: str) and a description. The body is fake.

Replace the body

Swap in a real backend. This example uses Tavily, but the pattern works for any HTTP API:

import os
import httpx
from bridle import tool

TAVILY_URL = "https://api.tavily.com/search"


@tool
def search(query: str) -> list[str]:
    """Search the web. Returns up to 10 result URLs.

    Use this when you need current information that's not in your training data.
    The query should be a focused phrase, not a question.
    """
    api_key = os.environ["TAVILY_API_KEY"]
    response = httpx.post(
        TAVILY_URL,
        json={"api_key": api_key, "query": query, "max_results": 10},
        timeout=15.0,
    )
    response.raise_for_status()
    payload = response.json()
    return [r["url"] for r in payload.get("results", [])]

Two things changed:

The body now hits a real backend.
The docstring got more specific — when to use it, what the query should look like. The model only ever sees the docstring, so it does the work of teaching the model how to use the tool.

Shape the output

The stub returns list[str]. That's fine when URLs are all the model needs, but a richer return often gives better downstream behavior:

from pydantic import BaseModel


class SearchHit(BaseModel):
    url: str
    title: str
    snippet: str


@tool
def search(query: str) -> list[SearchHit]:
    """Search the web. Returns up to 10 result hits with title and snippet."""
    api_key = os.environ["TAVILY_API_KEY"]
    response = httpx.post(
        TAVILY_URL,
        json={"api_key": api_key, "query": query, "max_results": 10},
        timeout=15.0,
    )
    response.raise_for_status()
    payload = response.json()
    return [
        SearchHit(
            url=r["url"],
            title=r.get("title", ""),
            snippet=r.get("content", "")[:500],
        )
        for r in payload.get("results", [])
    ]

The model now sees title and snippet for each hit, so it can pick which URLs to chase without an extra round-trip.

By default, exceptions inside a tool become tool-result errors that the model sees and can recover from. This is right for "this query returned nothing — try a different one." It is wrong for "the API key is invalid; stop." Use raise_on_error=True when the failure should abort the step:

@tool(raise_on_error=True)
def search(query: str) -> list[SearchHit]:
    """..."""
    api_key = os.environ["TAVILY_API_KEY"]   # KeyError aborts immediately
    ...

For finer control, raise inside the body:

@tool
def search(query: str) -> list[SearchHit]:
    """..."""
    if not query.strip():
        raise ValueError("Empty query")  # model sees this, retries with a non-empty query
    response = httpx.post(...)
    response.raise_for_status()           # network error: model sees, retries
    ...

Cap external cost

External calls cost money or rate-limit budget. Cache aggressively when results are stable:

import bridle
from bridle import cache, step
from bridle.cache.file import FileCache

bridle.set_cache(FileCache("./.bridle-cache"))

# Wrap the step that uses the tool — cache key includes the tools' fingerprint.
sources = cache(
    step(
        "find sources on Mars weather",
        schema=list[SearchHit],
        tools=[search],
    ),
    ttl=24 * 60 * 60,  # one day
)

Note that the cache lives at the step boundary, not at the tool boundary. The same query inside two different steps still hits the API. If you want per-tool memoization independent of the step, layer your own LRU cache on the function:

from functools import lru_cache

@lru_cache(maxsize=512)
def _cached_search(query: str) -> tuple[SearchHit, ...]:
    return tuple(_real_search(query))


@tool
def search(query: str) -> list[SearchHit]:
    """..."""
    return list(_cached_search(query))

The @tool wrapper does not interfere — Tool.__call__ forwards to the function, including the LRU cache.

Test the tool without the model

A Tool is callable. Hit it directly:

def test_search_returns_hits():
    hits = search("solar wind on Mars")
    assert all(isinstance(h, SearchHit) for h in hits)
    assert all(h.url.startswith("http") for h in hits)

For tests that should not hit the network, mock httpx.post or wrap the tool with mock inside an integration test:

from bridle import mock

# In a test:
fake_step = mock(
    step("find sources", schema=list[SearchHit], tools=[search]),
    [SearchHit(url="https://example.com", title="Example", snippet="...")],
)
result = bridle.resolve(fake_step)

Drop the tool back into the agent

Wire the new search into the brief writer. No other changes:

from bridle import agent, branch, cache, loop, retry, step

@agent(input=str, output=Brief, model="claude-sonnet-4-6")
def brief_writer(topic: str) -> Brief:
    plan = cache(step("...", schema=Plan, context=topic))

    sources: list[SearchHit] = []
    for t in plan.topics:
        found = loop(
            f"gather distinct sources on {t.title}",
            schema=SearchHit,
            until=lambda acc: len(acc) >= 2,
            tools=[search],   # the real tool
            max_iterations=4,
        )
        sources.extend(found)

    ...

Adding a real tool

Start from the stub

Replace the body

Shape the output

Decide the error policy

Cap external cost

Test the tool without the model

Drop the tool back into the agent

Next

Tools

Caching

A research brief writer

On this page