Integration and APIs

REST APIs use HTTP methods (GET, POST, PUT, DELETE, PATCH) to perform operations on resources identified by URLs. Agent tool functions call these APIs and return the result as a dictionary that the agent can reason about. Use httpx rather than Python's older requests library: httpx is async-native, which matters when tool calls run concurrently in a multi-agent system.

Every HTTP client in a tool function needs an explicit timeout. Without a timeout, a slow or unresponsive API causes the agent loop to hang indefinitely, consuming the thread and blocking any queued requests. Set timeouts between 10 and 30 seconds depending on the expected response time of the API.

import httpx, os
from typing import Optional

BASE_URL = "https://api.example.com/v1"

async def get_customer(customer_id: str) -> dict:
    """Fetch customer details from the CRM API."""
    api_key = os.getenv("CRM_API_KEY")
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f"{BASE_URL}/customers/{customer_id}",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Accept": "application/json"
            }
        )
        response.raise_for_status()  # raises on 4xx/5xx
        return response.json()

GraphQL is a query language for APIs developed by Facebook (now Meta) that lets clients request exactly the data they need. Unlike REST, a single GraphQL endpoint handles all operations. This reduces over-fetching: instead of receiving a full customer object when you only need the email address, you specify the fields you want.

One GraphQL-specific behaviour catches many developers: GraphQL always returns HTTP 200, even when the query fails. A query with a syntax error, an invalid variable, or an authorisation failure returns HTTP 200 with an errors array in the response body. An agent tool that only checks the HTTP status code will incorrectly treat GraphQL failures as successes.

async def graphql_query(query: str, variables: Optional[dict] = None) -> dict:
    token = os.getenv("INTERNAL_API_TOKEN")
    async with httpx.AsyncClient(timeout=15.0) as client:
        response = await client.post(
            "https://api.example.com/graphql",
            json={"query": query, "variables": variables or {}},
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
        )
        response.raise_for_status()
        data = response.json()

        # GraphQL returns 200 even for errors; always check this field
        if "errors" in data:
            return {"error": "GraphQL errors", "details": data["errors"]}
        return data.get("data", {})

API key authentication is the simplest pattern: include the key in an HTTP header on every request. Always load keys from environment variables, never hardcode them in the tool function or the tool schema. The tool schema is visible to the model; anything hardcoded there becomes part of the context window.

OAuth 2.0 (defined in RFC 6749) is the standard for server-to-server authentication where your agent acts as a service. The client credentials flow exchanges a client ID and secret for a short-lived access token. The token expires; your implementation must refresh it before expiry, not after the first 401 response.

import time

class OAuthClient:
    def __init__(self, token_url: str, client_id: str, client_secret: str):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self._access_token: Optional[str] = None
        self._token_expires_at: float = 0

    async def _refresh_token(self) -> None:
        async with httpx.AsyncClient() as client:
            response = await client.post(self.token_url, data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "scope": "read write"
            })
            response.raise_for_status()
            token_data = response.json()
            self._access_token = token_data["access_token"]
            # Subtract 60 seconds buffer to refresh before expiry
            self._token_expires_at = time.time() + token_data["expires_in"] - 60

    async def get_token(self) -> str:
        if not self._access_token or time.time() > self._token_expires_at:
            await self._refresh_token()
        return self._access_token

External APIs enforce rate limits. An agent that retries immediately on a 429 (Too Many Requests) response will be blocked for longer, not shorter. Exponential backoff waits twice as long after each failed attempt: 1 second, then 2, then 4, then 8. This gives the rate limiter time to reset.

Many APIs include a Retry-After header on 429 responses that specifies exactly how long to wait. Always check for this header before calculating your own backoff delay. Adding a small random jitter (up to 1 second) to the delay prevents multiple concurrent agent calls from retrying simultaneously and re-triggering the rate limit.

import asyncio, random

async def api_call_with_retry(fn, *args, max_retries: int = 3, base_delay: float = 1.0, **kwargs) -> dict:
    """Retry on 429 and 503 with exponential backoff. Never retry on 4xx client errors."""
    for attempt in range(max_retries + 1):
        try:
            return await fn(*args, **kwargs)
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status in (429, 503) and attempt < max_retries:
                retry_after = e.response.headers.get("Retry-After")
                delay = float(retry_after) if retry_after else base_delay * (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(delay)
                continue
            raise  # re-raise non-retryable errors immediately
    raise RuntimeError(f"Exhausted {max_retries} retries")

The circuit breaker pattern stops sending requests to an API that is consistently failing. After a defined number of consecutive failures, the breaker opens and all requests to that API fail immediately without attempting the call. After a recovery period, the breaker moves to half-open and allows one test request through. If it succeeds, the breaker closes and normal operation resumes.

When an API call fails, the tool function must not propagate the raw exception to the agent. A raw Python exception message such as httpx.ConnectError: [Errno 111] Connection refused is not useful to the agent: it cannot reason about what to do next. Catch exceptions at the tool boundary and return a structured dictionary with a human-readable, actionable error description.

Include a success boolean so the agent can check the result programmatically rather than parsing the message text. Map common HTTP status codes to plain-language descriptions that the agent can relay to the user without exposing internal system details.

async def safe_api_call(endpoint: str, params: dict) -> dict:
    try:
        result = await api_call_with_retry(make_request, endpoint, params)
        return {"success": True, "data": result}
    except httpx.HTTPStatusError as e:
        return {
            "success": False,
            "error_type": "http_error",
            "status_code": e.response.status_code,
            "message": {
                400: "Invalid request parameters. Check the input and try again.",
                401: "Authentication failed. The API key may be invalid or expired.",
                403: "Access denied. This operation requires additional permissions.",
                404: "Resource not found. The requested item does not exist.",
                429: "Rate limit reached. The system will retry automatically.",
                500: "The external service is experiencing issues. Try again later."
            }.get(e.response.status_code, f"HTTP error {e.response.status_code}")
        }
    except httpx.TimeoutException:
        return {
            "success": False,
            "error_type": "timeout",
            "message": "The request timed out. The service may be slow or unavailable."
        }