Connect an AI Agent to an API

Connect an AI agent to an API through an application-owned adapter: validate the proposed tool and arguments, check access, call a fixed HTTP operation with application credentials, and return a small validated result or a controlled error. The model should propose the operation; it should not choose the credential or unrestricted destination URL.

This walkthrough builds a read-only stock lookup against a local fixture API. It covers timeouts, malformed JSON, 401, 429 and server failures, with at most two HTTP attempts. No live model, real inventory service or secret is required.

An adapter translates one interface into another. Here, it turns a normalized tool request into an HTTP GET and translates the response back into data the host can safely inspect. It does not decide whether the agent should make the request in the first place.

Separate the model call from HTTP execution

Imagine an illustrative shop assistant answering whether SKU-100 is available. A SKU is a product identifier. The application exposes a narrow get_stock tool rather than a general “fetch any URL” capability.

A normalized request for this example is:

{"name": "get_stock", "arguments": {"sku": "SKU-100"}}

This is the example’s local format, not a copy-paste request to a model provider. In a real host, your model SDK receives a tool call, you parse its arguments and map it into this shape, and the host associates the returned result with the original call ID. OpenAI’s function-calling guide documents that separation between a model request, application execution and the returned observation.

The host must also authorize the user’s access before execution. A correctly shaped SKU is not proof of permission, and a service credential is not a substitute for user-specific access checks. Our fixture contains public synthetic stock only, so it does not implement a customer authorization system.

Define a small failure and retry policy

Retries need a policy, not just a loop. HTTP semantics distinguish idempotent requests, whose intended effect can be repeated, from operations that cannot safely be retried without additional knowledge. This example retries only a read-only GET and only selected responses.

RFC 6585 defines 429 as rate limiting and permits a Retry-After header. That header can use seconds or an HTTP date under RFC 9110. This small implementation accepts integer seconds up to one second. It deliberately defers longer waits or unsupported dates rather than shortening the server’s requested delay.

The same Retry-After handling applies when a selected server error supplies that header. The two-attempt cap and one-second retry-wait allowance are example choices. A real integration needs a total deadline, coordinated retry budgets and usually jitter to spread concurrent retries. Do not copy this policy onto a payment, email or other write operation.

Build the validated read-only adapter

The three files below use only the standard library and were run with Python 3.8.20. Save them together in an empty tutorial directory. There is no package installation or change to your application’s environment.

Save stock_adapter.py:

import http.client
import json
import re
import socket
import time


def request_stock(port, sku, token):
    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=0.05)
    try:
        connection.request(
            "GET", "/stock/" + sku,
            headers={"Authorization": "Bearer " + token},
        )
        response = connection.getresponse()
        body = response.read(4097)
        if len(body) > 4096:
            raise ValueError("Response too large")
        return response.status, response.getheader("Retry-After"), body
    finally:
        connection.close()


def execute(call, port, token, send=request_stock, sleep=time.sleep):
    def result(code, attempts, data=None):
        output = {"ok": code == "ok", "code": code, "attempts": attempts}
        if data is not None:
            output["data"] = data
        return output

    if not isinstance(call, dict) or set(call) != {"name", "arguments"}:
        return result("invalid_call", 0)
    args = call["arguments"]
    if call["name"] != "get_stock" or not isinstance(args, dict):
        return result("invalid_call", 0)
    if set(args) != {"sku"} or not isinstance(args["sku"], str):
        return result("invalid_arguments", 0)
    sku = args["sku"]
    if not re.fullmatch(r"SKU-[0-9]{3}", sku):
        return result("invalid_arguments", 0)
    if not isinstance(token, str) or not re.fullmatch(r"[A-Za-z0-9._-]{1,128}", token):
        return result("missing_credentials", 0)

    for attempt in (1, 2):
        try:
            status, retry_after, body = send(port, sku, token)
        except socket.timeout:
            return result("timeout", attempt)
        except (OSError, http.client.HTTPException):
            return result("network_error", attempt)
        except ValueError:
            return result("bad_response", attempt)
        if status == 401:
            return result("unauthorized", attempt)
        if status == 404:
            return result("not_found", attempt)
        if status in (429, 502, 503, 504):
            code = "rate_limited" if status == 429 else "upstream_error"
            delay = 0.1
            if status == 429 or retry_after is not None:
                if retry_after is None or not re.fullmatch(r"[0-9]+", retry_after):
                    return result(code, attempt)
                delay = int(retry_after) if len(retry_after) <= 3 else 999
            if attempt == 2 or delay > 1:
                return result(code, attempt)
            sleep(delay)
            continue
        if status != 200:
            return result("upstream_error" if status >= 500 else "upstream_rejected", attempt)
        try:
            data = json.loads(body.decode("utf-8"))
        except (ValueError, UnicodeError):
            return result("bad_response", attempt)
        if not isinstance(data, dict) or set(data) != {"sku", "available"}:
            return result("bad_response", attempt)
        if data["sku"] != sku or type(data["available"]) is not int:
            return result("bad_response", attempt)
        if not 0 <= data["available"] <= 10000:
            return result("bad_response", attempt)
        return result("ok", attempt, data)

The tool accepts exactly one SKU argument and rejects extra fields, including a model-supplied token or URL. The application supplies the port and credential separately. The transport uses only 127.0.0.1, sends GET, follows no redirects and closes each connection.

The 50-millisecond socket timeout makes the local failure test quick. Python’s versioned HTTP client documentation describes this timeout for blocking operations. It is not an end-to-end deadline: multiple reads, retries and waiting can increase total elapsed time.

Response reads are capped at 4 KiB. Successful JSON must have exactly sku and available, match the requested SKU, and contain an integer from 0 to 10,000. Booleans are rejected as quantities even though Python treats bool as a subclass of int. This is a narrow response contract, not a general JSON validator.

Errors return fixed codes without raw response bodies, exception details, request headers or credentials. The adapter does not convert a failed lookup into a claim that stock is zero.

Supply a local API that can fail on demand

Save fixture_api.py:

import json
import time
from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Lock

GOOD = json.dumps({"sku": "SKU-100", "available": 12}).encode("utf-8")
FAKE_TOKEN = "fixture-only-key"


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        with self.server.lock:
            self.server.calls += 1
            reply = self.server.replies.popleft() if self.server.replies else (503, None, b"")
        if self.headers.get("Authorization") != "Bearer " + FAKE_TOKEN:
            reply = (401, None, b"")
        elif self.path != "/stock/SKU-100":
            reply = (404, None, b"")
        status, retry_after, body = reply
        if status == "slow":
            time.sleep(0.2)
            status, body = 200, GOOD
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            if retry_after is not None:
                self.send_header("Retry-After", retry_after)
            self.end_headers()
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            pass

    def log_message(self, *args):
        pass


def make_api(replies):
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    server.replies = deque(replies)
    server.calls = 0
    server.lock = Lock()
    return server

This fixture binds to loopback on an available port and serves scripted replies. The public string fixture-only-key is intentionally fake; it is not a real secret or a pattern for hardcoding production credentials. The fixture’s simple header comparison only helps exercise the 401 path.

For a real service, keep the credential in application configuration or a secret manager, outside model arguments and conversation logs. Use HTTPS with certificate verification and a configured destination allowlist. Do not send a real token through this HTTP-only demo. Check the provider’s authentication, scopes and token-refresh rules separately.

The slow fixture response can outlive a client timeout. Its broken-pipe handling is expected test behavior, not evidence that a timeout stopped the server’s work. The standard-library server is a test fixture, not a production deployment recommendation.

Test policy decisions and real HTTP separately

Save check_adapter.py:

import json
import os
import socket
from threading import Thread

from fixture_api import FAKE_TOKEN, GOOD, make_api
from stock_adapter import execute

CALL = {"name": "get_stock", "arguments": {"sku": "SKU-100"}}


def check(label, replies, expected, attempts, token=FAKE_TOKEN):
    server = make_api(replies)
    worker = Thread(target=server.serve_forever)
    worker.start()
    waits = []
    try:
        answer = execute(CALL, server.server_port, token, sleep=waits.append)
        assert answer["code"] == expected and answer["attempts"] == attempts
        assert server.calls == attempts
        if token:
            assert token not in json.dumps(answer)
        print(label + ":", answer["code"], "attempts=" + str(attempts))
        return answer
    finally:
        server.shutdown()
        server.server_close()
        worker.join()


def policy_tests():
    calls, waits = [], []

    def transient(*args):
        calls.append("GET")
        return (503, None, b"") if len(calls) == 1 else (200, None, GOOD)

    answer = execute(CALL, 0, FAKE_TOKEN, send=transient, sleep=waits.append)
    assert answer["ok"] and calls == ["GET", "GET"] and waits == [0.1]

    def timed_out(*args):
        raise socket.timeout()

    assert execute(CALL, 0, FAKE_TOKEN, send=timed_out)["code"] == "timeout"

    def forbidden(*args):
        raise AssertionError("Invalid input must not reach HTTP")

    invalid = {"name": "get_stock", "arguments": {"sku": "../../private"}}
    assert execute(invalid, 0, FAKE_TOKEN, send=forbidden)["attempts"] == 0
    print("deterministic policy: passed")


if __name__ == "__main__":
    policy_tests()
    token = os.environ.get("STOCK_DEMO_TOKEN", FAKE_TOKEN)
    check("success", [(200, None, GOOD)], "ok", 1, token)
    check("malformed JSON", [(200, None, b"not-json")], "bad_response", 1)
    check("401", [(200, None, GOOD)], "unauthorized", 1, "wrong-fixture-key")
    check("429 then success", [(429, "0", b""), (200, None, GOOD)], "ok", 2)
    check("long Retry-After", [(429, "999", b"")], "rate_limited", 1)
    check("503 then success", [(503, None, b""), (200, None, GOOD)], "ok", 2)
    check("persistent 503", [(503, None, b""), (503, None, b"")], "upstream_error", 2)
    check("500", [(500, None, b"")], "upstream_error", 1)
    check("timeout", [("slow", None, b"")], "timeout", 1)

The deterministic tests inject a transport and a wait recorder. They prove that a transient response gets one retry with the expected delay, a timeout becomes a controlled result, and invalid input does not reach HTTP. No clock timing is needed for those checks.

The remaining cases use the actual local HTTP fixture. They replace sleeping with waits.append to record requested delays without waiting. That verifies retry decisions, not real backoff timing. The timeout case separately exercises an actual slow socket response.

Run from the directory containing all three files:

python3.8 check_adapter.py

Expected stdout, with exit status 0:

deterministic policy: passed
success: ok attempts=1
malformed JSON: bad_response attempts=1
401: unauthorized attempts=1
429 then success: ok attempts=2
long Retry-After: rate_limited attempts=1
503 then success: ok attempts=2
persistent 503: upstream_error attempts=2
500: upstream_error attempts=1
timeout: timeout attempts=1

The optional STOCK_DEMO_TOKEN environment variable is read by the test driver; leave it unset or use the public fixture value. No real key is needed. The other cases deliberately use the fixture credential or an explicitly wrong fake value.

A successful adapter result has this shape:

{"ok": true, "code": "ok", "attempts": 1, "data": {"sku": "SKU-100", "available": 12}}

The host can return that observation to the model through its SDK, preserving the provider’s call ID and result format. If ok is false, it should explain the failure or arrange a handoff. The code above does not call a model, measure model judgment or demonstrate a complete agent loop.

Carry the boundaries into the real integration

Keep input validation, user authorization and credentials in the host/application layer. Treat API responses as untrusted data, even when authentication succeeds. A provider can return HTML, malformed JSON, a different SKU or an error message containing sensitive details.

Before replacing the fixture, test the provider’s exact schema and status behavior. Add a total deadline, response/content-type policy, appropriate JSON parsing rules, rate-limit coordination and redacted audit records. This example does not reject every possible duplicate-key JSON representation and does not implement a universal HTTP client.

The advantage of a narrow adapter is that it exposes one useful business operation while keeping HTTP failures and secrets out of model-facing inputs. The cost is maintenance: schema changes, credential rotation, provider limits and retry behavior all need an owner.

For a fixed report that always requests the same SKU, a script or ordinary API client may be simpler than an agent. Use a model when choosing the next lookup or explaining ambiguous evidence adds value. For that separate decision, see AI Agents vs Chatbots vs Workflows.

Start with one read-only operation and its failure cases. Keep the deterministic adapter tests when you add the model SDK, then evaluate the complete interaction separately. A reliable HTTP adapter gives the agent useful observations; it does not guarantee that the agent will make the right decision.

Topics: AI agents, HTTP APIs, tool adapters, input validation, timeouts, rate limits, retries, secret handling, Python.