Agent integration

Give an LLM agent a tool that hits paid APIs. Routeweiler handles the 402 transparently from inside the tool call.

Prerequisites

Requirement	Notes
Agent framework	OpenAI Agents SDK shown; LangChain, Anthropic tool use, and MCP follow the same closure pattern
Funding source	Any rail will work. This recipe uses x402 on Base Sepolia. See x402 — Base USDC for wallet setup
A paid endpoint	Use the x402 test server for local testing
OpenAI API key	Required for the worked example; not needed for other frameworks

Environment variables

export WALLET_KEY=0x<your-private-key>
export OPENAI_API_KEY=sk-...
export WEATHER_URL=http://127.0.0.1:4021/forecast

The pattern

Create the Routeweiler client once in the outer async with block and let each tool function close over it. Every tool invocation then shares the same funding source, budget envelope, and trace sink, so cost accounting and policy enforcement are automatic across the full agent run.

LLM → tool call → rw.get(url) → 402 Payment Required
                                       ↓
                               Routeweiler pays
                                       ↓
                               200 OK + receipt
                  ← tool returns text
LLM continues

The agent never sees the payment. It only sees the 200 response body.

Worked example: OpenAI Agents SDK

import asyncio
import os

from agents import Agent, Runner, function_tool
from eth_account import Account

from routeweiler import (
    BudgetExceededError,
    Funding,
    NoFeasibleRailError,
    PolicyDeniedError,
    Routeweiler,
)
from routeweiler.trace.sink_sqlite import TraceSink

WEATHER_URL = os.environ["WEATHER_URL"]


async def main() -> None:
    wallet = Account.from_key(os.environ["WALLET_KEY"])

    async with Routeweiler(
        funding=[Funding.base_sepolia_usdc(wallet=wallet)],
        trace_sink=TraceSink.sqlite("rw.db"),
        agent_id="weather-agent",
    ) as rw:

        @function_tool
        async def fetch_weather(city: str) -> str:
            """Fetch the current weather for a city from a paid weather API."""
            try:
                resp = await rw.get(f"{WEATHER_URL}?city={city}")
                return resp.text
            except BudgetExceededError:
                return "Budget cap reached — no more weather lookups this session."
            except PolicyDeniedError as exc:
                return f"Policy blocked the request: {exc}"
            except NoFeasibleRailError:
                return "No usable payment rail — check your funding source."

        agent = Agent(
            name="Weather Assistant",
            model="gpt-4o-mini",
            instructions="You help users check weather. Use the fetch_weather tool when asked.",
            tools=[fetch_weather],
        )

        result = await Runner.run(agent, "What is the current weather in London?")
        print(result.final_output)


asyncio.run(main())

Returning error strings (rather than raising) lets the model react and report back to the user gracefully. Raised exceptions inside a @function_tool abort the entire run.

Adding a budget envelope

When an LLM drives tool invocations there is no per-call human review, and the model may call the same tool repeatedly. A budget envelope caps total spend across the full run:

from routeweiler import BudgetEnvelope

async with Routeweiler(
    funding=[Funding.base_sepolia_usdc(wallet=wallet)],
    trace_sink=TraceSink.sqlite("rw.db"),
    agent_id="weather-agent",
    budget_envelope=BudgetEnvelope(
        id="weather-agent-session",
        cap_minor_units=100,        # $1.00 in cents
        cap_currency="usd",
        allowed_rails=["x402"],
        ttl_seconds=3_600,
    ),
) as rw:
    ...

Once the cap is hit, BudgetExceededError is raised on the next rw.get(...) call. The try/except inside the tool catches it and returns a string so the model can acknowledge the limit without crashing the run.

Server side (for local testing)

rw-demo/demo6_agent/server.py is a minimal FastAPI server that wraps GET /forecast with PaymentMiddlewareASGI and charges $0.0001 in Base-Sepolia USDC. Run it alongside the agent:

export ROUTEWEILER_TEST_MERCHANT_RECIPIENT=0x<your-address>
python rw-demo/demo6_agent/server.py &

See x402 — Test endpoint for a fuller explanation of the server-side setup.

Pattern variants

The async with Routeweiler(...) as rw block and the tool closure work the same way in other frameworks:

Anthropic tool use

Define the tool schema as a dict, run the model loop yourself, and dispatch on tool_use content blocks. The tool handler body is await rw.get(url). Wrap the same async with Routeweiler(...) context around the loop.

LangChain

from langchain_core.tools import tool

@tool
async def fetch_weather(city: str) -> str:
    """Fetch weather from a paid API."""
    resp = await rw.get(f"{WEATHER_URL}?city={city}")
    return resp.text

Create rw before the agent executor and keep it alive for the duration of the run.

MCP server

Expose rw.get as an MCP tool so any MCP client (Claude Desktop, Claude Code) can use your agent's wallet without knowing anything about x402 or Routeweiler:

@server.call_tool()
async def fetch_weather(name: str, arguments: dict) -> list:
    resp = await rw.get(f"{WEATHER_URL}?city={arguments['city']}")
    return [TextContent(type="text", text=resp.text)]

What to check in the trace

Each tool-call invocation appends one row to trace_events in rw.db, keyed by agent_id:

import sqlite3, json

conn = sqlite3.connect("rw.db")
conn.row_factory = sqlite3.Row
rows = conn.execute(
    "SELECT * FROM trace_events WHERE agent_id = 'weather-agent'"
).fetchall()
conn.close()

for row in rows:
    payload = json.loads(row["payload"])
    payment = payload["payment"]
    print(f"Tx:        {payment['proofValue']}")
    print(f"Amount:    {payment['amountNative']} base units = {payment['amountEnvelope']} USD")
    print(f"Settled:   {payment['settlementLatencyMs']} ms")

This gives you a per-tool-call audit trail for the entire agent run.

Error handling

Inside each tool, return errors as strings so the model can react:

from routeweiler import (
    BudgetExceededError,
    PolicyDeniedError,
    NoFeasibleRailError,
    RouteweilerError,
)

@function_tool
async def fetch_weather(city: str) -> str:
    try:
        resp = await rw.get(f"{WEATHER_URL}?city={city}")
        return resp.text
    except BudgetExceededError:
        return "Budget cap reached — no more payments this session."
    except PolicyDeniedError as exc:
        return f"Policy blocked the request: {exc}"
    except NoFeasibleRailError:
        return "No usable rail — check funding source or policy."

Outside the agent run, catch RouteweilerError to handle unexpected failures without leaking internal details:

from routeweiler import RouteweilerError

try:
    result = await Runner.run(agent, prompt)
except RouteweilerError as exc:
    print(f"Payment infrastructure error: {exc}")

See also

• x402 — Base USDC
• x402 — Test endpoint
• Concepts: Budgets
• Concepts: Traces
• OpenAI Agents SDK docs