Wrapping a REST API as an Agent Tool

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