Custom Tools Basics

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"

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,
        },
    },
}

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"),
    }

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)

# 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]

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.

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}

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)

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)