API Integration

Connecting agents to external APIs: patterns, authentication, resilience, and best practices.

API Integration Patterns

1. Direct HTTP Client

import httpx

class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0,
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
        )

    async def get(self, path: str, params: dict = None):
        response = await self.client.get(path, params=params)
        response.raise_for_status()
        return response.json()

    async def post(self, path: str, data: dict):
        response = await self.client.post(path, json=data)
        response.raise_for_status()
        return response.json()

2. OpenAPI/Swagger Client Generation

# Generate client from OpenAPI spec
pip install openapi-python-client
openapi-python-client generate --path openapi.json --output ./client

# Generated client
from my_api_client import Client
from my_api_client.api.users import get_user

client = Client(base_url="https://api.example.com", headers={"Authorization": "Bearer token"})
user = await get_user.asyncio(client=client, user_id="123")

3. GraphQL Client

import strawberry
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport

transport = AIOHTTPTransport(
    url="https://api.example.com/graphql",
    headers={"Authorization": "Bearer token"}
)

client = Client(transport=transport, fetch_schema_from_transport=True)

query = gql("""
    query GetUser($id: ID!) {
        user(id: $id) {
            id
            name
            email
            posts { title }
        }
    }
""")

result = await client.execute_async(query, variable_values={"id": "123"})

4. MCP Server as API Gateway

# MCP server wraps external APIs
@server.tool()
async def github_create_issue(
    repo: str, 
    title: str, 
    body: str,
    labels: list[str] = []
) -> TextContent:
    """Create a GitHub issue via GitHub API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://api.github.com/repos/{repo}/issues",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            json={"title": title, "body": body, "labels": labels}
        )
        response.raise_for_status()
        return TextContent(type="text", text=str(response.json()))

Authentication Patterns

1. API Key

# Header
headers = {"X-API-Key": api_key}
# or
headers = {"Authorization": f"Bearer {api_key}"}

2. OAuth 2.0 (Client Credentials)

class OAuthClient:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self._token = None
        self._expires_at = 0

    async def get_token(self) -> str:
        if time.time() < self._expires_at - 60:
            return self._token

        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
                }
            )
            data = response.json()
            self._token = data["access_token"]
            self._expires_at = time.time() + data["expires_in"]
            return self._token

3. OAuth 2.0 (Authorization Code + PKCE)

# For user-facing agents
class UserOAuthClient:
    def __init__(self, client_id: str, redirect_uri: str):
        self.client_id = client_id
        self.redirect_uri = redirect_uri
        self.code_verifier = self.generate_pkce()

    def get_auth_url(self) -> str:
        challenge = self.pkce_challenge(self.code_verifier)
        return f"https://auth.example.com/authorize?" + urlencode({
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "response_type": "code",
            "scope": "read write",
            "code_challenge": challenge,
            "code_challenge_method": "S256",
            "state": self.generate_state()
        })

    async def exchange_code(self, code: str) -> TokenResponse:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://auth.example.com/token",
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "redirect_uri": self.redirect_uri,
                    "client_id": self.client_id,
                    "code_verifier": self.code_verifier
                }
            )
            return response.json()

4. mTLS (Mutual TLS)

ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain("client.pem", "client.key")
ssl_context.load_verify_locations("ca.pem")

async with httpx.AsyncClient(verify=ssl_context) as client:
    response = await client.get("https://api.example.com/data")

Resilience Patterns

1. Retry with Exponential Backoff

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class ResilientClient:
    def __init__(self):
        self.client = httpx.AsyncClient()

    @retry(
        wait=wait_exponential(multiplier=1, min=1, max=10),
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError, httpx.HTTPStatusError)),
        before_sleep=lambda retry_state: logger.warning(f"Retrying: {retry_state.outcome.exception()}")
    )
    async def get(self, url: str):
        response = await self.client.get(url)
        response.raise_for_status()
        return response.json()

2. Circuit Breaker

from circuitbreaker import circuit

class CircuitBreakerClient:
    def __init__(self):
        self.client = httpx.AsyncClient()

    @circuit(failure_threshold=5, recovery_timeout=30, expected_exception=httpx.HTTPError)
    async def call_api(self, method: str, url: str, **kwargs):
        response = await getattr(self.client, method)(url, **kwargs)
        response.raise_for_status()
        return response.json()

3. Timeout & Bulkhead

class ConfiguredClient:
    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=5.0,      # Connection timeout
                read=30.0,        # Read timeout
                write=10.0,       # Write timeout
                pool=5.0          # Pool timeout
            ),
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
                keepalive_expiry=30.0
            )
        )

    # Per-request timeout override
    async def get_with_timeout(self, url: str, timeout: float = 10.0):
        return await self.client.get(url, timeout=timeout)

4. Idempotency Keys

class IdempotentClient:
    def __init__(self, client: httpx.AsyncClient):
        self.client = client

    async def post_idempotent(self, url: str, data: dict, idempotency_key: str):
        headers = {"Idempotency-Key": idempotency_key}
        response = await self.client.post(url, json=data, headers=headers)
        response.raise_for_status()
        return response.json()

Rate Limiting

Client-Side

from asyncio import Semaphore
from collections import defaultdict
import time

class RateLimitedClient:
    def __init__(self, client: httpx.AsyncClient, max_per_second: int = 10):
        self.client = client
        self.semaphore = Semaphore(max_per_second)
        self.requests = defaultdict(list)

    async def request(self, method: str, url: str, **kwargs):
        async with self.semaphore:
            # Token bucket per host
            host = httpx.URL(url).host
            now = time.time()
            self.requests[host] = [t for t in self.requests[host] if now - t < 1.0]

            if len(self.requests[host]) >= 10:
                await asyncio.sleep(0.1)

            self.requests[host].append(now)
            return await getattr(self.client, method.lower())(url)

Server-Side (Respect Headers)

async def handle_rate_limit(response: httpx.Response):
    if response.status_code == 429:
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            wait = int(retry_after)
        else:
            wait = 60

        logger.warning(f"Rate limited, waiting {wait}s")
        await asyncio.sleep(wait)
        return True  # Retry
    return False

Error Handling

Structured Errors

class APIError(Exception):
    def __init__(self, status_code: int, message: str, response: httpx.Response = None):
        self.status_code = status_code
        self.message = message
        self.response = response
        super().__init__(f"API Error {status_code}: {message}")

class APIClient:
    async def request(self, method: str, url: str, **kwargs) -> dict:
        try:
            response = await self.client.request(method, url, **kwargs)

            if response.status_code >= 400:
                # Parse error details
                try:
                    error_data = response.json()
                    message = error_data.get("message", error_data.get("error", "Unknown error"))
                except:
                    message = response.text

                raise APIError(response.status_code, message, response)

            return response.json()

        except httpx.TimeoutException:
            raise APIError(0, "Request timeout")
        except httpx.ConnectError:
            raise APIError(0, "Connection failed")

Error Classification

def classify_error(error: APIError) -> ErrorClass:
    if error.status_code == 0:
        return ErrorClass.NETWORK
    elif error.status_code == 401:
        return ErrorClass.AUTH
    elif error.status_code == 403:
        return ErrorClass.FORBIDDEN
    elif error.status_code == 404:
        return ErrorClass.NOT_FOUND
    elif error.status_code == 429:
        return ErrorClass.RATE_LIMIT
    elif 500 <= error.status_code < 600:
        return ErrorClass.SERVER
    else:
        return ErrorClass.CLIENT

SDK Generation & Management

OpenAPI Client Generation

# Generate typed client
openapi-python-client generate \
    --path https://api.example.com/openapi.json \
    --output ./generated_client \
    --package-name my_api_client \
    --overwrite

Version Pinning

# requirements.txt
my-api-client==1.2.3  # Pin exact version

# Or with pip-tools
# requirements.in
my-api-client>=1.2.0,<2.0.0

Deprecation Handling

class DeprecationHandler:
    def __init__(self, client):
        self.client = client
        self.deprecation_warnings = set()

    async def request(self, method, url, **kwargs):
        response = await self.client.request(method, url, **kwargs)

        # Check for deprecation headers
        if "Deprecation" in response.headers:
            warn_key = f"{url}:{response.headers['Deprecation']}"
            if warn_key not in self.deprecation_warnings:
                logger.warning(f"API deprecated: {url} - {response.headers['Deprecation']}")
                self.deprecation_warnings.add(warn_key)

        if "Sunset" in response.headers:
            sunset_date = response.headers["Sunset"]
            logger.warning(f"API sunset: {url} on {sunset_date}")

        return response

Testing API Integrations

Mock Server

import respx
import httpx

@respx.mock
async def test_api_call():
    # Mock response
    respx.get("https://api.example.com/users/123").mock(
        return_value=httpx.Response(200, json={"id": "123", "name": "Test"})
    )

    client = APIClient()
    user = await client.get_user("123")
    assert user["name"] == "Test"

Contract Testing (Pact)

# Consumer test
def test_user_api_contract(pact):
    pact.given("user exists").upon_receiving("request for user").with_request(
        method="GET", path="/users/123"
    ).will_respond_with(
        status=200,
        body=like({"id": "123", "name": "Test", "email": "test@example.com"})
    )

    with pact:
        client = APIClient(base_url=pact.uri)
        user = client.get_user("123")
        assert user["name"] == "Test"

Integration Testing

@pytest.mark.integration
async def test_real_api():
    client = APIClient(base_url="https://api.staging.example.com")

    # Test with real API
    user = await client.get_user("test-user-123")
    assert user["email"] == "test@example.com"

    # Cleanup
    await client.delete_user("test-user-123")

Monitoring & Observability

Request Logging

class LoggingClient:
    def __init__(self, client: httpx.AsyncClient):
        self.client = client

    async def request(self, method: str, url: str, **kwargs):
        start = time.time()
        request_id = str(uuid4())

        logger.info("API Request", extra={
            "request_id": request_id,
            "method": method,
            "url": url,
            "headers": self._sanitize_headers(kwargs.get("headers", {}))
        })

        try:
            response = await self.client.request(method, url, **kwargs)
            duration = time.time() - start

            logger.info("API Response", extra={
                "request_id": request_id,
                "status_code": response.status_code,
                "duration_ms": duration * 1000
            })

            return response
        except Exception as e:
            logger.error("API Error", extra={
                "request_id": request_id,
                "error": str(e),
                "duration_ms": (time.time() - start) * 1000
            })
            raise

Metrics

from prometheus_client import Histogram, Counter

API_REQUEST_DURATION = Histogram(
    "api_request_duration_seconds",
    "API request latency",
    ["method", "endpoint", "status"]
)

API_REQUESTS_TOTAL = Counter(
    "api_requests_total",
    "Total API requests",
    ["method", "endpoint", "status"]
)

class MetricsClient:
    async def request(self, method: str, url: str, **kwargs):
        start = time.time()
        try:
            response = await self.client.request(method, url, **kwargs)
            status = response.status_code
            return response
        except Exception as e:
            status = 0
            raise
        finally:
            duration = time.time() - start
            endpoint = httpx.URL(url).path
            API_REQUEST_DURATION.labels(method, endpoint, status).observe(duration)
            API_REQUESTS_TOTAL.labels(method, endpoint, status).inc()

Function Calling — API as tool
MCP Protocol — API via MCP
Code Execution — Code calling APIs
Browser Automation — Browser vs API
Custom Tools — Building API tools
Tool Discovery — API discovery
Secure Coding — API security
Supply Chain Security — API supply chain
CI CD Best Practices — API in CI/CD
Observability — API observability
Security — API security
Observability — API observability
Reliability — Circuit breakers, retries
Cost Optimization — API cost optimization

Sources: httpx docs, OpenAPI spec, GraphQL spec, OAuth 2.0 RFC, tenacity, circuitbreaker, respx, Pact, Prometheus, Stripe API patterns, GitHub API patterns, AWS SDK patterns, Google Cloud API patterns.

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown