REST API Tool

Use codemap find "RestAPITool" after adding the pattern to your project.

REST API Invocation Tool with Auth Propagation

An httpx-based @tool() class that supports all common HTTP methods and propagates authorization headers from the calling request context.

Critical rule — name override for acronym classes

RestAPITool would auto-generate rest_a_p_i_tool. Override explicitly:

@tool(name="rest_api_tool")
class RestAPITool:
    ...

Pattern

from lauren_ai._tools import tool, ToolContext
import httpx
import json

@tool(name="rest_api_tool")
class RestAPITool:
    """Invoke a REST API endpoint.

    Args:
        url: The full URL to call (or path when base_url is set).
        method: HTTP method (GET, POST, PUT, DELETE).
        body: Optional JSON body for POST/PUT as a JSON string.
        headers: Optional additional headers as a JSON string.
    """

    def __init__(self, base_url: str = "", auth_header: str | None = None):
        self._base_url = base_url
        self._auth_header = auth_header

    async def run(
        self,
        ctx: ToolContext,
        url: str,
        method: str = "GET",
        body: str = "",
        headers: str = "",
    ) -> dict:
        full_url = self._base_url + url if url.startswith("/") else url
        extra_headers = json.loads(headers) if headers else {}

        # Static auth header from tool configuration
        if self._auth_header:
            extra_headers["Authorization"] = self._auth_header

        # Dynamic auth propagation from the originating request
        if ctx.execution_context and ctx.execution_context.request:
            auth = ctx.execution_context.request.headers.get("authorization")
            if auth:
                extra_headers["Authorization"] = auth

        async with httpx.AsyncClient(timeout=30.0) as client:
            try:
                method_fn = getattr(client, method.lower())
                response = await method_fn(
                    full_url,
                    json=json.loads(body) if body else None,
                    headers=extra_headers,
                )
                return {"status": response.status_code, "body": response.text[:2000]}
            except Exception as e:
                return {"error": str(e)}

AgentRunner test pattern

Use unittest.mock.patch to intercept httpx.AsyncClient within the tool call's context. Pass the tool instance to @use_tools and use _Capture to inspect results.

import json
from unittest.mock import AsyncMock, MagicMock, patch
from lauren_ai._agents import AgentContext, agent, use_tools
from lauren_ai._tools import ToolResult
from lauren_ai._transport import Completion, TokenUsage
from lauren_ai.testing import TestClient


class _Capture:
    def __init__(self):
        self.captured: list[ToolResult] = []

    async def on_tool_result(self, result: ToolResult, ctx: AgentContext) -> ToolResult | None:
        self.captured.append(result)
        return None


def _c(text):
    return Completion(id="c1", model="mock", content=text, tool_calls=[],
                      stop_reason="end_turn", usage=TokenUsage(10, 5))


def _make_agent(tool_instance):
    @agent(model=None, system="REST API test agent")
    @use_tools(tool_instance)
    class RestAPITestAgent(_Capture):
        def __init__(self):
            _Capture.__init__(self)
    return RestAPITestAgent()


def test_get_request_is_made():
    tool_inst = RestAPITool()
    agent_inst = _make_agent(tool_inst)
    client = TestClient(agent_inst)

    mock_resp = MagicMock(status_code=200, text='{"ok": true}')
    mock_client = MagicMock()
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    mock_client.get = AsyncMock(return_value=mock_resp)

    with patch("httpx.AsyncClient", return_value=mock_client):
        client.mock.queue_tool_use(
            "rest_api_tool",
            {"url": "https://api.example.com/data", "method": "GET"},
        )
        client.mock.queue_response(_c("Data retrieved."))
        client.run("Fetch data")

    data = json.loads(agent_inst.captured[0].content)
    assert data["status"] == 200

Auth propagation chain

HTTP request ─► Lauren controller
                    │ execution_context.request.headers["authorization"]
                    ▼
            AgentRunner.run(..., execution_context=ctx)
                    │
                    ▼
            ToolContext.execution_context
                    │
                    ▼
            RestAPITool.run → httpx call with Authorization header

Notes

• httpx.AsyncClient is created per-call. For high-throughput use, inject a shared client via the constructor (DI-friendly).
• body and headers are JSON strings so they can be passed through the LLM tool-call JSON encoding.
• Cap response.text[:2000] to avoid flooding the model context.
• Always override the tool name with @tool(name="rest_api_tool") since the auto-generated name for RestAPITool would be rest_a_p_i_tool.

Source: lauren-framework/lauren-ai — distributed by TomeVault.