Wrapping a REST API as an Agent Tool

Q: Should I expose pagination cursors?

Yes for large sets. Return next_cursor and require it on the next call instead of huge pages.

Wrapping REST means building a thin, validated host function that exposes one agent-facing action while your code owns auth, timeouts, error mapping, and response shaping.

You do not hand the model a general HTTP client. You hand it a named capability.

Summary

Pick a domain action, design a strict JSON schema, implement a handler that calls REST with scoped credentials, return compact structured results, and register the tool in your agent runtime.

Recipe

Choose one job (for example list_open_incidents, not call_any_path).
Write a model-facing description that states when to call and what not to call it for.
Define JSON Schema parameters with types, enums, limits, and additionalProperties: false.
Map parameters to the REST path, query, and body (never let the model supply raw URLs or headers).
Inject auth from the host secret store; scope the token to that action.
Set timeouts, retries only for safe methods or idempotent writes, and max response size.
Normalize success and failure into a small JSON object (ok, data fields, error_code, retryable).
Unit-test the handler without a model; then eval the tool inside an agent task suite.
Version the tool name if the contract changes in a breaking way.

Working Example

Wrap GET /v1/incidents?status=&limit= as list_incidents.

import json
import os
from typing import Any, Literal
 
import httpx
from pydantic import BaseModel, Field, ValidationError
 
BASE = os.environ["INCIDENTS_API_BASE"]  # e.g. https://api.internal.example
TOKEN = os.environ["INCIDENTS_READ_TOKEN"]
 
class ListIncidentsArgs(BaseModel):
    status: Literal["open", "acked", "resolved"] = "open"
    limit: int = Field(default=10, ge=1, le=50)
    team: str | None = Field(default=None, max_length=64)
 
TOOL_SPEC = {
    "type": "function",
    "function": {
        "name": "list_incidents",
        "description": (
            "List recent incidents filtered by status. "
            "Use to survey workload before acknowledging or escalating. "
            "Do not use for creating or resolving incidents."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["open", "acked", "resolved"],
                    "description": "Incident lifecycle filter.",
                },
                "limit": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 50,
                    "description": "Max rows to return (default 10).",
                },
                "team": {
                    "type": "string",
                    "description": "Optional team slug filter.",
                },
            },
            "additionalProperties": False,
        },
    },
}
 
def list_incidents(arguments_json: str) -> str:
    try:
        args = ListIncidentsArgs.model_validate_json(arguments_json or "{}")
    except ValidationError as e:
        return json.dumps({
            "ok": False,
            "error_code": "INVALID_ARG",
            "message": e.errors(),
            "retryable": False,
        })
 
    params: dict[str, Any] = {"status": args.status, "limit": args.limit}
    if args.team:
        params["team"] = args.team
 
    try:
        with httpx.Client(timeout=httpx.Timeout(10.0, connect=5.0)) as client:
            r = client.get(
                f"{BASE}/v1/incidents",
                params=params,
                headers={"Authorization": f"Bearer {TOKEN}", "Accept": "application/json"},
            )
    except httpx.TimeoutException:
        return json.dumps({"ok": False, "error_code": "TIMEOUT", "retryable": True})
    except httpx.HTTPError:
        return json.dumps({"ok": False, "error_code": "NETWORK", "retryable": True})
 
    if r.status_code == 401 or r.status_code == 403:
        return json.dumps({"ok": False, "error_code": "AUTH", "message": "Not allowed", "retryable": False})
    if r.status_code == 429:
        return json.dumps({"ok": False, "error_code": "RATE_LIMIT", "retryable": True})
    if r.status_code >= 500:
        return json.dumps({"ok": False, "error_code": "UPSTREAM", "retryable": True})
    if r.status_code >= 400:
        return json.dumps({"ok": False, "error_code": "BAD_REQUEST", "message": r.text[:300], "retryable": False})
 
    payload = r.json()
    items = payload.get("items") or payload.get("data") or []
    slim = [
        {
            "id": i.get("id"),
            "title": (i.get("title") or "")[:120],
            "severity": i.get("severity"),
            "status": i.get("status"),
            "opened_at": i.get("opened_at"),
        }
        for i in items[: args.limit]
    ]
    return json.dumps({"ok": True, "count": len(slim), "incidents": slim})

Deep Dive

Mapping REST to tool surface

REST concept	Agent tool choice
Resource collection GET	`list_*` with filters + limit
Resource GET by id	`get_*` requiring id from list/search
POST create	`create_*` with idempotency key
PATCH/PUT	`update_*` with explicit fields only
DELETE	Separate tool + approval for high impact
Batch / search DSL	Fixed query templates, not free strings

Collapse multi-call UI flows when the agent should not invent intermediate states (for example one refund_order tool that calls validate + refund endpoints server-side).

Auth and tenancy

Prefer service credentials scoped to the tool's verbs.
Propagate end-user identity as a signed internal header your API trusts, not as a model-supplied token.
For multi-tenant SaaS, bind tenant id from the session, never from free tool args alone.

Retries

Method	Default retry in tool host
GET / HEAD	Safe with backoff on 429/5xx/timeout
POST with Idempotency-Key	Safe if backend honors the key
POST without key	Do not auto-retry
PATCH/DELETE	Retry only if documented idempotent

Surface retryable to the model, but enforce max attempts in the host so the loop cannot hammer an outage.

Response shaping

Drop fields the model does not need for the next decision.
Cap list length and string length.
Include ids required by follow-up tools.
Prefer ISO timestamps and enums over prose status blobs.

Observability

Log tool name, arg hashes (or redacted args), latency, status, upstream request id, and agent run id.

Never log full secrets or unnecessary PII.

Gotchas

URL-as-argument tools. If the model can pass any path or host, you built an SSRF gadget.
Passing through error HTML. Models thrash on pages; map to codes.
Oversized JSON. A 200-item customer dump will crowd out instructions.
Shared superuser tokens. One injected prompt then has admin powers.
Silent 204 / empty bodies. Return an explicit {ok: true, empty: true} style result.
Schema drift. Backend renames a field while the tool description still promises the old shape.
Framework double-parsing. Some SDKs pass dict args already; handle both string and dict inputs carefully.

Alternatives

Approach	Pros	Cons
Hand-written tool per action	Safest, clearest	More code
OpenAPI → generated tools	Fast coverage	Often too wide / noisy
MCP server over curated tools	Multi-client reuse	Still need curation
Generic `http_request` tool	Flexible	High abuse and SSRF risk
GraphQL free-form tool	Flexible reads	Over-fetch and auth holes

FAQs

How many REST endpoints should one tool cover?

Usually one primary action. Internal fan-out is fine if the agent sees one semantic step.

Should I expose pagination cursors?

Yes for large sets. Return next_cursor and require it on the next call instead of huge pages.

What timeout should I use?

Start with connect 2-5s and total 10-30s depending on the API. Long jobs should return a job id and a status tool.

How do I wrap authenticated user OAuth APIs?

Store OAuth tokens server-side per user/session. The tool uses the session principal; the model never sees the refresh token.

Can I stream REST responses into the model?

Most tool APIs expect a finished string result. Buffer with a size cap, or write large artifacts to object storage and return a pointer.

What if the REST API is only XML?

Translate to JSON in the handler. Keep the model on one structured format.

Do I need separate tools for staging and production?

Prefer one tool name with environment-specific base URLs and credentials in config, plus hard network isolation between envs.

How should write tools signal success?

Return the new resource id, version/etag, and a short status. That enables follow-up gets without re-querying by fuzzy text.

Related: Custom Tools Basics

Related: What Makes an Internal API a Good Agent Tool

Related: Versioning and Deprecating Agent Tools Without Breaking Agents

Related: Scoping API Keys and Credentials Passed to an Agent

Related: Custom Tools Best Practices

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

You do not hand the model a general HTTP client. You hand it a named capability.

Summary

Pick a domain action, design a strict JSON schema, implement a handler that calls REST with scoped credentials, return compact structured results, and register the tool in your agent runtime.

Recipe

Choose one job (for example list_open_incidents, not call_any_path).
Write a model-facing description that states when to call and what not to call it for.
Define JSON Schema parameters with types, enums, limits, and additionalProperties: false.
Map parameters to the REST path, query, and body (never let the model supply raw URLs or headers).
Inject auth from the host secret store; scope the token to that action.
Set timeouts, retries only for safe methods or idempotent writes, and max response size.
Normalize success and failure into a small JSON object (ok, data fields, error_code, retryable).
Unit-test the handler without a model; then eval the tool inside an agent task suite.
Version the tool name if the contract changes in a breaking way.

Working Example

Wrap GET /v1/incidents?status=&limit= as list_incidents.

import json
import os
from typing import Any, Literal
 
import httpx
from pydantic import BaseModel, Field, ValidationError
 
BASE = os.environ["INCIDENTS_API_BASE"]  # e.g. https://api.internal.example
TOKEN = os.environ["INCIDENTS_READ_TOKEN"]
 
class ListIncidentsArgs(BaseModel):
    status: Literal["open", "acked", "resolved"] = "open"
    limit: int = Field(default=10, ge=1, le=50)
    team: str | None = Field(default=None, max_length=64)
 
TOOL_SPEC = {
    "type": "function",
    "function": {
        "name": "list_incidents",
        "description": (
            "List recent incidents filtered by status. "
            "Use to survey workload before acknowledging or escalating. "
            "Do not use for creating or resolving incidents."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["open", "acked", "resolved"],
                    "description": "Incident lifecycle filter.",
                },
                "limit": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 50,
                    "description": "Max rows to return (default 10).",
                },
                "team": {
                    "type": "string",
                    "description": "Optional team slug filter.",
                },
            },
            "additionalProperties": False,
        },
    },
}
 
def list_incidents(arguments_json: str) -> str:
    try:
        args = ListIncidentsArgs.model_validate_json(arguments_json or "{}")
    except ValidationError as e:
        return json.dumps({
            "ok": False,
            "error_code": "INVALID_ARG",
            "message": e.errors(),
            "retryable": False,
        })
 
    params: dict[str, Any] = {"status": args.status, "limit": args.limit}
    if args.team:
        params["team"] = args.team
 
    try:
        with httpx.Client(timeout=httpx.Timeout(10.0, connect=5.0)) as client:
            r = client.get(
                f"{BASE}/v1/incidents",
                params=params,
                headers={"Authorization": f"Bearer {TOKEN}", "Accept": "application/json"},
            )
    except httpx.TimeoutException:
        return json.dumps({"ok": False, "error_code": "TIMEOUT", "retryable": True})
    except httpx.HTTPError:
        return json.dumps({"ok": False, "error_code": "NETWORK", "retryable": True})
 
    if r.status_code == 401 or r.status_code == 403:
        return json.dumps({"ok": False, "error_code": "AUTH", "message": "Not allowed", "retryable": False})
    if r.status_code == 429:
        return json.dumps({"ok": False, "error_code": "RATE_LIMIT", "retryable": True})
    if r.status_code >= 500:
        return json.dumps({"ok": False, "error_code": "UPSTREAM", "retryable": True})
    if r.status_code >= 400:
        return json.dumps({"ok": False, "error_code": "BAD_REQUEST", "message": r.text[:300], "retryable": False})
 
    payload = r.json()
    items = payload.get("items") or payload.get("data") or []
    slim = [
        {
            "id": i.get("id"),
            "title": (i.get("title") or "")[:120],
            "severity": i.get("severity"),
            "status": i.get("status"),
            "opened_at": i.get("opened_at"),
        }
        for i in items[: args.limit]
    ]
    return json.dumps({"ok": True, "count": len(slim), "incidents": slim})

Deep Dive

Mapping REST to tool surface

REST concept	Agent tool choice
Resource collection GET	`list_*` with filters + limit
Resource GET by id	`get_*` requiring id from list/search
POST create	`create_*` with idempotency key
PATCH/PUT	`update_*` with explicit fields only
DELETE	Separate tool + approval for high impact
Batch / search DSL	Fixed query templates, not free strings

Collapse multi-call UI flows when the agent should not invent intermediate states (for example one refund_order tool that calls validate + refund endpoints server-side).

Auth and tenancy

Prefer service credentials scoped to the tool's verbs.
Propagate end-user identity as a signed internal header your API trusts, not as a model-supplied token.
For multi-tenant SaaS, bind tenant id from the session, never from free tool args alone.

Retries

Method	Default retry in tool host
GET / HEAD	Safe with backoff on 429/5xx/timeout
POST with Idempotency-Key	Safe if backend honors the key
POST without key	Do not auto-retry
PATCH/DELETE	Retry only if documented idempotent

Surface retryable to the model, but enforce max attempts in the host so the loop cannot hammer an outage.

Response shaping

Drop fields the model does not need for the next decision.
Cap list length and string length.
Include ids required by follow-up tools.
Prefer ISO timestamps and enums over prose status blobs.

Observability

Log tool name, arg hashes (or redacted args), latency, status, upstream request id, and agent run id.

Never log full secrets or unnecessary PII.

Gotchas

URL-as-argument tools. If the model can pass any path or host, you built an SSRF gadget.
Passing through error HTML. Models thrash on pages; map to codes.
Oversized JSON. A 200-item customer dump will crowd out instructions.
Shared superuser tokens. One injected prompt then has admin powers.
Silent 204 / empty bodies. Return an explicit {ok: true, empty: true} style result.
Schema drift. Backend renames a field while the tool description still promises the old shape.
Framework double-parsing. Some SDKs pass dict args already; handle both string and dict inputs carefully.

Alternatives

Approach	Pros	Cons
Hand-written tool per action	Safest, clearest	More code
OpenAPI → generated tools	Fast coverage	Often too wide / noisy
MCP server over curated tools	Multi-client reuse	Still need curation
Generic `http_request` tool	Flexible	High abuse and SSRF risk
GraphQL free-form tool	Flexible reads	Over-fetch and auth holes

FAQs

How many REST endpoints should one tool cover?

Usually one primary action. Internal fan-out is fine if the agent sees one semantic step.

Should I expose pagination cursors?

Yes for large sets. Return next_cursor and require it on the next call instead of huge pages.

What timeout should I use?

Start with connect 2-5s and total 10-30s depending on the API. Long jobs should return a job id and a status tool.

How do I wrap authenticated user OAuth APIs?

Store OAuth tokens server-side per user/session. The tool uses the session principal; the model never sees the refresh token.

Can I stream REST responses into the model?

Most tool APIs expect a finished string result. Buffer with a size cap, or write large artifacts to object storage and return a pointer.

What if the REST API is only XML?

Translate to JSON in the handler. Keep the model on one structured format.

Do I need separate tools for staging and production?

Prefer one tool name with environment-specific base URLs and credentials in config, plus hard network isolation between envs.

How should write tools signal success?

Return the new resource id, version/etag, and a short status. That enables follow-up gets without re-querying by fuzzy text.

Related: Custom Tools Basics

Related: What Makes an Internal API a Good Agent Tool

Related: Versioning and Deprecating Agent Tools Without Breaking Agents

Related: Scoping API Keys and Credentials Passed to an Agent

Related: Custom Tools Best Practices

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Wrapping a REST API as an Agent Tool

Search Documentation

Wrapping a REST API as an Agent Tool