# API Integration

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

---

## API Integration Patterns

### 1. Direct HTTP Client
```python
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
```bash
# Generate client from OpenAPI spec
pip install openapi-python-client
openapi-python-client generate --path openapi.json --output ./client
```

```python
# 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
```python
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
```python
# 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
```python
# Header
headers = {"X-API-Key": api_key}
# or
headers = {"Authorization": f"Bearer {api_key}"}
```

### 2. OAuth 2.0 (Client Credentials)
```python
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)
```python
# 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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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)
```python
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
```python
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
```python
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
```bash
# 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
```python
# 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
```python
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
```python
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)
```python
# 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
```python
@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
```python
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
```python
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()
```

---

## Related Notes

- [[ToolUse/Function-Calling.md]] — API as tool
- [[ToolUse/MCP-Protocol.md]] — API via MCP
- [[ToolUse/Code-Execution.md]] — Code calling APIs
- [[ToolUse/Browser-Automation.md]] — Browser vs API
- [[ToolUse/Custom-Tools.md]] — Building API tools
- [[ToolUse/Tool-Discovery.md]] — API discovery
- [[Security/Secure-Coding.md]] — API security
- [[Security/Supply-Chain-Security.md]] — API supply chain
- [[DevOps/CI-CD-Best-Practices.md]] — API in CI/CD
- [[DevOps/Observability.md]] — API observability
- [[Production/Security.md]] — API security
- [[Production/Observability.md]] — API observability
- [[Production/Reliability.md]] — Circuit breakers, retries
- [[Production/Cost-Optimization.md]] — 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.*