Custom Tools Basics

8 examples to get you started with custom agent tools - 5 basic and 3 intermediate.

You will define a schema, call a mock internal REST API, return structured results, handle errors, and wire the tool into a minimal call-and-respond loop.

Prerequisites

Python 3.11+ recommended
Familiarity with JSON and HTTP status codes
Optional: an LLM API key if you run the final agent loop against a real model

python -m venv .venv && source .venv/bin/activate
pip install httpx pydantic openai
export OPENAI_API_KEY="..."          # or OPENROUTER_API_KEY + base_url
export INTERNAL_API_BASE="http://127.0.0.1:8000"

These examples use plain Python so the pattern ports to LangGraph, Pydantic AI, OpenAI Agents SDK, and others.

Basic Examples

1. Define a Tool Schema the Model Can Call

Start with a name, description, and JSON Schema parameters - the contract the model sees.

GET_ORDER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_order",
        "description": (
            "Fetch one order by numeric order_id. "
            "Use after the user provides an order number. "
            "Returns status, total_cents, and currency."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "integer",
                    "minimum": 1,
                    "description": "Positive order id from the user or a prior list_orders call.",
                }
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
    },
}

Descriptions tell the model when to call; property descriptions tell it how.
additionalProperties: False reduces invented fields.
Keep one clear job per tool name.

Related: What Makes an Internal API a Good Agent Tool

2. Implement the Host-Side Handler

The model never hits your API directly. Your code validates args and calls REST.

import os
import httpx
 
BASE = os.environ.get("INTERNAL_API_BASE", "http://127.0.0.1:8000")
 
def get_order(order_id: int) -> dict:
    if order_id < 1:
        return {"ok": False, "error_code": "INVALID_ARG", "message": "order_id must be >= 1", "retryable": False}
    with httpx.Client(timeout=10.0) as client:
        r = client.get(f"{BASE}/orders/{order_id}", headers={"Authorization": f"Bearer {os.environ.get('INTERNAL_API_TOKEN', '')}"})
    if r.status_code == 404:
        return {"ok": False, "error_code": "NOT_FOUND", "message": f"Order {order_id} not found", "retryable": False}
    if r.status_code >= 500:
        return {"ok": False, "error_code": "UPSTREAM", "message": "Order service unavailable", "retryable": True}
    r.raise_for_status()
    data = r.json()
    return {
        "ok": True,
        "order_id": data["id"],
        "status": data["status"],
        "total_cents": data["total_cents"],
        "currency": data.get("currency", "USD"),
    }

Map HTTP outcomes to a small JSON shape the model can reason about.
Keep secrets in the host environment, never in tool args.
Timeouts belong in the client, not "hope the API is fast."

3. Dispatch Tool Calls by Name

When the model returns a tool call, look up the handler and pass parsed JSON.

import json
 
HANDLERS = {
    "get_order": lambda args: get_order(int(args["order_id"])),
}
 
def run_tool(name: str, arguments_json: str) -> str:
    args = json.loads(arguments_json or "{}")
    handler = HANDLERS.get(name)
    if handler is None:
        return json.dumps({"ok": False, "error_code": "UNKNOWN_TOOL", "message": name, "retryable": False})
    try:
        result = handler(args)
    except Exception as exc:  # log exc server-side
        result = {"ok": False, "error_code": "HANDLER", "message": "Tool failed", "retryable": False}
    return json.dumps(result)

Always return a string (JSON text) to the model API.
Catch unexpected exceptions so the agent loop can continue or escalate.
Unknown tool names should fail closed.

4. Mock the Internal API for Local Practice

Run a tiny FastAPI (or any) stub so tools work offline.

# mock_orders.py - run: uvicorn mock_orders:app --port 8000
from fastapi import FastAPI, HTTPException
 
app = FastAPI()
ORDERS = {
    42: {"id": 42, "status": "shipped", "total_cents": 2599, "currency": "USD"},
}
 
@app.get("/orders/{order_id}")
def order(order_id: int):
    if order_id not in ORDERS:
        raise HTTPException(status_code=404, detail="not found")
    return ORDERS[order_id]

Local mocks unlock unit tests without production credentials.
Seed edge cases: 404, 500, slow responses.
Swap INTERNAL_API_BASE when you move to staging.

5. Return Compact Success Payloads

Prefer fields the agent needs next, not the full database row.

def summarize_order(raw: dict) -> dict:
    return {
        "ok": True,
        "order_id": raw["id"],
        "status": raw["status"],
        "total_cents": raw["total_cents"],
        "currency": raw.get("currency", "USD"),
    }
# Avoid shipping PII, internal notes, or multi-KB debug blobs to the model.

Smaller results leave room for reasoning and later tool calls.
Redact PII unless the product truly requires it in context.
Stable keys beat clever nested shapes.

Related: Wrapping a REST API as an Agent Tool

Intermediate Examples

6. Add a Write Tool With an Idempotency Key

Creates need safe retries when the agent loop repeats.

import uuid
import httpx
 
def create_support_ticket(subject: str, body: str, idempotency_key: str | None = None) -> dict:
    key = idempotency_key or str(uuid.uuid4())
    with httpx.Client(timeout=15.0) as client:
        r = client.post(
            f"{BASE}/tickets",
            headers={
                "Authorization": f"Bearer {os.environ['INTERNAL_API_TOKEN']}",
                "Idempotency-Key": key,
                "Content-Type": "application/json",
            },
            json={"subject": subject[:200], "body": body[:4000]},
        )
    if r.status_code >= 500:
        return {"ok": False, "error_code": "UPSTREAM", "retryable": True, "idempotency_key": key}
    if r.status_code >= 400:
        return {"ok": False, "error_code": "BAD_REQUEST", "message": r.text[:300], "retryable": False}
    data = r.json()
    return {"ok": True, "ticket_id": data["id"], "idempotency_key": key}

Pass the same key on retry so the backend dedupes.
Truncate free-text fields so prompts cannot explode payloads.
Tell the model in the tool description when to create vs search first.

7. Minimal Model Loop (One Tool Round-Trip)

Wire schema + handler into a single assistant turn with tools.

import json
import os
from openai import OpenAI
 
client = OpenAI()  # set base_url for OpenRouter if needed
tools = [GET_ORDER_TOOL]
 
messages = [{"role": "user", "content": "What is the status of order 42?"}]
resp = client.chat.completions.create(
    model="gpt-4o-mini",  # verify slug at build
    messages=messages,
    tools=tools,
)
msg = resp.choices[0].message
messages.append(msg)
 
if msg.tool_calls:
    for tc in msg.tool_calls:
        payload = run_tool(tc.function.name, tc.function.arguments)
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": payload,
        })
    final = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools)
    print(final.choices[0].message.content)
else:
    print(msg.content)

Frameworks hide this loop; understanding it clarifies bugs.
Always append tool results with the matching tool_call_id.
Cap outer iterations in production agents.

Related: Tool Use & Function Calling Basics

8. Validate Arguments With Pydantic Before I/O

Schema in the prompt is soft. Enforce hard validation in the host.

from pydantic import BaseModel, Field, ValidationError
 
class GetOrderArgs(BaseModel):
    order_id: int = Field(ge=1)
 
def get_order_validated(arguments_json: str) -> dict:
    try:
        args = GetOrderArgs.model_validate_json(arguments_json)
    except ValidationError as e:
        return {"ok": False, "error_code": "INVALID_ARG", "message": e.errors(), "retryable": False}
    return get_order(args.order_id)

Double validation catches model drift and prompt injection junk.
Return field-level errors so the model can self-correct.
Keep validation errors small (no full stack traces).

What to Practice Next

Wrap a real staging REST endpoint with the same pattern.
Add read-only vs write tools with different credentials.
Version tool names when schemas change.

Related: Web Search and Browsing Tools for Research Agents

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.

Custom Tools Basics

8 examples to get you started with custom agent tools - 5 basic and 3 intermediate.

You will define a schema, call a mock internal REST API, return structured results, handle errors, and wire the tool into a minimal call-and-respond loop.

Prerequisites

Python 3.11+ recommended
Familiarity with JSON and HTTP status codes
Optional: an LLM API key if you run the final agent loop against a real model

python -m venv .venv && source .venv/bin/activate
pip install httpx pydantic openai
export OPENAI_API_KEY="..."          # or OPENROUTER_API_KEY + base_url
export INTERNAL_API_BASE="http://127.0.0.1:8000"

These examples use plain Python so the pattern ports to LangGraph, Pydantic AI, OpenAI Agents SDK, and others.

Basic Examples

1. Define a Tool Schema the Model Can Call

Start with a name, description, and JSON Schema parameters - the contract the model sees.

GET_ORDER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_order",
        "description": (
            "Fetch one order by numeric order_id. "
            "Use after the user provides an order number. "
            "Returns status, total_cents, and currency."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "integer",
                    "minimum": 1,
                    "description": "Positive order id from the user or a prior list_orders call.",
                }
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
    },
}

Descriptions tell the model when to call; property descriptions tell it how.
additionalProperties: False reduces invented fields.
Keep one clear job per tool name.

Related: What Makes an Internal API a Good Agent Tool

2. Implement the Host-Side Handler

The model never hits your API directly. Your code validates args and calls REST.

import os
import httpx
 
BASE = os.environ.get("INTERNAL_API_BASE", "http://127.0.0.1:8000")
 
def get_order(order_id: int) -> dict:
    if order_id < 1:
        return {"ok": False, "error_code": "INVALID_ARG", "message": "order_id must be >= 1", "retryable": False}
    with httpx.Client(timeout=10.0) as client:
        r = client.get(f"{BASE}/orders/{order_id}", headers={"Authorization": f"Bearer {os.environ.get('INTERNAL_API_TOKEN', '')}"})
    if r.status_code == 404:
        return {"ok": False, "error_code": "NOT_FOUND", "message": f"Order {order_id} not found", "retryable": False}
    if r.status_code >= 500:
        return {"ok": False, "error_code": "UPSTREAM", "message": "Order service unavailable", "retryable": True}
    r.raise_for_status()
    data = r.json()
    return {
        "ok": True,
        "order_id": data["id"],
        "status": data["status"],
        "total_cents": data["total_cents"],
        "currency": data.get("currency", "USD"),
    }

Map HTTP outcomes to a small JSON shape the model can reason about.
Keep secrets in the host environment, never in tool args.
Timeouts belong in the client, not "hope the API is fast."

3. Dispatch Tool Calls by Name

When the model returns a tool call, look up the handler and pass parsed JSON.

import json
 
HANDLERS = {
    "get_order": lambda args: get_order(int(args["order_id"])),
}
 
def run_tool(name: str, arguments_json: str) -> str:
    args = json.loads(arguments_json or "{}")
    handler = HANDLERS.get(name)
    if handler is None:
        return json.dumps({"ok": False, "error_code": "UNKNOWN_TOOL", "message": name, "retryable": False})
    try:
        result = handler(args)
    except Exception as exc:  # log exc server-side
        result = {"ok": False, "error_code": "HANDLER", "message": "Tool failed", "retryable": False}
    return json.dumps(result)

Always return a string (JSON text) to the model API.
Catch unexpected exceptions so the agent loop can continue or escalate.
Unknown tool names should fail closed.

4. Mock the Internal API for Local Practice

Run a tiny FastAPI (or any) stub so tools work offline.

# mock_orders.py - run: uvicorn mock_orders:app --port 8000
from fastapi import FastAPI, HTTPException
 
app = FastAPI()
ORDERS = {
    42: {"id": 42, "status": "shipped", "total_cents": 2599, "currency": "USD"},
}
 
@app.get("/orders/{order_id}")
def order(order_id: int):
    if order_id not in ORDERS:
        raise HTTPException(status_code=404, detail="not found")
    return ORDERS[order_id]

Local mocks unlock unit tests without production credentials.
Seed edge cases: 404, 500, slow responses.
Swap INTERNAL_API_BASE when you move to staging.

5. Return Compact Success Payloads

Prefer fields the agent needs next, not the full database row.

def summarize_order(raw: dict) -> dict:
    return {
        "ok": True,
        "order_id": raw["id"],
        "status": raw["status"],
        "total_cents": raw["total_cents"],
        "currency": raw.get("currency", "USD"),
    }
# Avoid shipping PII, internal notes, or multi-KB debug blobs to the model.

Smaller results leave room for reasoning and later tool calls.
Redact PII unless the product truly requires it in context.
Stable keys beat clever nested shapes.

Related: Wrapping a REST API as an Agent Tool

Intermediate Examples

6. Add a Write Tool With an Idempotency Key

Creates need safe retries when the agent loop repeats.

import uuid
import httpx
 
def create_support_ticket(subject: str, body: str, idempotency_key: str | None = None) -> dict:
    key = idempotency_key or str(uuid.uuid4())
    with httpx.Client(timeout=15.0) as client:
        r = client.post(
            f"{BASE}/tickets",
            headers={
                "Authorization": f"Bearer {os.environ['INTERNAL_API_TOKEN']}",
                "Idempotency-Key": key,
                "Content-Type": "application/json",
            },
            json={"subject": subject[:200], "body": body[:4000]},
        )
    if r.status_code >= 500:
        return {"ok": False, "error_code": "UPSTREAM", "retryable": True, "idempotency_key": key}
    if r.status_code >= 400:
        return {"ok": False, "error_code": "BAD_REQUEST", "message": r.text[:300], "retryable": False}
    data = r.json()
    return {"ok": True, "ticket_id": data["id"], "idempotency_key": key}

Pass the same key on retry so the backend dedupes.
Truncate free-text fields so prompts cannot explode payloads.
Tell the model in the tool description when to create vs search first.

7. Minimal Model Loop (One Tool Round-Trip)

Wire schema + handler into a single assistant turn with tools.

import json
import os
from openai import OpenAI
 
client = OpenAI()  # set base_url for OpenRouter if needed
tools = [GET_ORDER_TOOL]
 
messages = [{"role": "user", "content": "What is the status of order 42?"}]
resp = client.chat.completions.create(
    model="gpt-4o-mini",  # verify slug at build
    messages=messages,
    tools=tools,
)
msg = resp.choices[0].message
messages.append(msg)
 
if msg.tool_calls:
    for tc in msg.tool_calls:
        payload = run_tool(tc.function.name, tc.function.arguments)
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": payload,
        })
    final = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools)
    print(final.choices[0].message.content)
else:
    print(msg.content)

Frameworks hide this loop; understanding it clarifies bugs.
Always append tool results with the matching tool_call_id.
Cap outer iterations in production agents.

Related: Tool Use & Function Calling Basics

8. Validate Arguments With Pydantic Before I/O

Schema in the prompt is soft. Enforce hard validation in the host.

from pydantic import BaseModel, Field, ValidationError
 
class GetOrderArgs(BaseModel):
    order_id: int = Field(ge=1)
 
def get_order_validated(arguments_json: str) -> dict:
    try:
        args = GetOrderArgs.model_validate_json(arguments_json)
    except ValidationError as e:
        return {"ok": False, "error_code": "INVALID_ARG", "message": e.errors(), "retryable": False}
    return get_order(args.order_id)

Double validation catches model drift and prompt injection junk.
Return field-level errors so the model can self-correct.
Keep validation errors small (no full stack traces).

What to Practice Next

Wrap a real staging REST endpoint with the same pattern.
Add read-only vs write tools with different credentials.
Version tool names when schemas change.

Related: Web Search and Browsing Tools for Research Agents

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.

Custom Tools Basics

Buscar en la documentación

Custom Tools Basics