Connecting Agents to External Systems

Why integration matters

Agents are only useful when they can talk to the real world: databases, APIs, email systems, payment processors. But real-world integrations are messy.

• Authentication fails: API keys expire, credentials aren't passed correctly, or external systems demand OAuth flows.
• Services are slow: A database query takes 10 seconds. An API times out. The agent can't wait forever.
• Services fail: The payment processor is down 1% of the time. The agent must recover gracefully instead of crashing.
• Rate limits exist: An API allows 100 requests per minute. The agent might hit that limit and need to backoff.

Integration patterns protect agents from these hazards.

Integration reliability: Common API characteristics

The table below summarizes typical SLA and performance characteristics of different integration types (illustrative estimates based on industry standards):

| Integration type | Typical uptime SLA | P50 latency | P99 latency | Rate limit (typical) | Retry strategy | |------------------|-------------------|------------|------------|----------------------|-----------------| | Payment processor (Stripe, Square) | 99.99% | 200ms | 1s | 100 req/s | Exponential backoff (3x retry) | | Cloud database (RDS, Cloud SQL) | 99.95% | 50ms | 200ms | 500-1000 conn | Connection pooling, circuit breaker | | REST API (public SaaS) | 99.5% | 300ms | 2s | 60-300 req/min | Retry with jitter, hourly quota | | Webhook delivery (incoming events) | 95% (best effort) | N/A (async) | Minutes | N/A | Exponential backoff (24h retry window) | | Legacy SOAP/XML service | 99% (often degraded) | 500ms+ | 5s+ | 10-50 req/min | Conservative retry (2x) |

Key insight: Payment processors and databases are more reliable but latency-sensitive. Public APIs and webhooks are less reliable and require aggressive retry logic. Legacy systems are slowest but also most likely to fail under load.

Authenticated API wrapper

Basic pattern with retries

import os
import time
import logging
from typing import Optional

logger = logging.getLogger(__name__)

class APIClient:
    """Wrapper for external API calls with auth, retry, and logging."""

    def __init__(self, api_key: Optional[str] = None, base_url: str = "https://api.example.com"):
        # Get API key from environment or parameter
        self.api_key = api_key or os.environ.get("EXTERNAL_API_KEY")
        if not self.api_key:
            raise ValueError("API key not found. Set EXTERNAL_API_KEY env var.")

        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "User-Agent": "AgentFramework/1.0"
        })

    def call(
        self,
        endpoint: str,
        method: str = "GET",
        params: dict = None,
        json_body: dict = None,
        max_retries: int = 3,
        timeout: int = 10
    ) -> dict:
        """
        Call the external API with automatic retry on transient failures.
        Returns the parsed JSON response or raises an error.
        """
        url = f"{self.base_url}/{endpoint}"
        last_error = None

        for attempt in range(max_retries):
            try:
                logger.debug(f"API call to {endpoint} (attempt {attempt + 1}/{max_retries})")

                response = self.session.request(
                    method=method,
                    url=url,
                    params=params,
                    json=json_body,
                    timeout=timeout
                )

                # Success: return parsed response
                if response.status_code == 200:
                    logger.info(f"API call succeeded: {endpoint}")
                    return response.json()

                # Client error: don't retry
                if response.status_code < 500:
                    error_msg = f"API error {response.status_code}: {response.text}"
                    logger.error(error_msg)
                    raise ValueError(error_msg)

                # Server error: retry with backoff
                logger.warning(f"API server error {response.status_code}, retrying...")
                last_error = response

            except requests.exceptions.Timeout:
                logger.warning(f"API call timed out, retrying...")
                last_error = TimeoutError("API call timed out")

            except requests.exceptions.ConnectionError as e:
                logger.warning(f"Connection error, retrying...")
                last_error = e

            except Exception as e:
                # Unexpected error: don't retry
                logger.error(f"Unexpected error: {e}")
                raise

            # Wait before retrying (exponential backoff with jitter)
            if attempt < max_retries - 1:
                wait_time = (2 ** attempt) + (random.random() * 0.1)  # Backoff + jitter
                logger.debug(f"Waiting {wait_time:.2f}s before retry...")
                time.sleep(wait_time)

        # All retries exhausted
        logger.error(f"API call failed after {max_retries} attempts")
        raise RuntimeError(f"API call to {endpoint} failed: {last_error}")

# Usage in agent:
api = APIClient()

# In a tool definition:
def fetch_customer_data(customer_id: str) -> dict:
    """Fetch customer details from external CRM."""
    try:
        result = api.call(
            endpoint=f"customers/{customer_id}",
            method="GET",
            timeout=5
        )
        return result
    except (ValueError, RuntimeError) as e:
        # Return error to agent so it can retry or escalate
        return {"error": str(e), "customer_id": customer_id}

Key features:

• Automatic retry: Server errors (5xx) trigger exponential backoff. Client errors (4xx) fail immediately.
• Jitter: Random delay prevents thundering herd (all retries happening at the same time).
• Logging: Every attempt is logged for debugging.
• Timeout: Prevents hanging forever waiting for a slow API.

Rate limiting and queuing

If the external API has strict rate limits, queue requests:

from queue import Queue
from threading import Thread
import time

class RateLimitedAPIClient:
    """API client that respects rate limits."""

    def __init__(self, api_key: str, requests_per_minute: int = 60):
        self.api_key = api_key
        self.requests_per_minute = requests_per_minute
        self.last_request_time = 0
        self.min_interval = 60.0 / requests_per_minute  # Seconds between requests

    def call(self, endpoint: str, **kwargs) -> dict:
        """Call API, enforcing rate limit."""
        # Wait until enough time has passed since last request
        elapsed = time.time() - self.last_request_time
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)

        # Now make the call
        self.last_request_time = time.time()
        response = self._make_request(endpoint, **kwargs)
        return response

    def _make_request(self, endpoint: str, **kwargs) -> dict:
        # Implementation similar to APIClient above
        pass

# Usage:
rate_limited_api = RateLimitedAPIClient(api_key="...", requests_per_minute=30)

# Agent can call this freely; the client enforces the rate limit
result = rate_limited_api.call("search", query="...")

Database integration

Agents often need to query databases. Here's a safe pattern:

import sqlite3
from contextlib import contextmanager

class DatabaseAgent:
    """Query a database with query validation."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        # Define allowed tables and columns to prevent SQL injection
        self.allowed_tables = {"users", "products", "orders"}
        self.allowed_columns = {
            "users": {"id", "name", "email", "created_at"},
            "products": {"id", "name", "price", "category"},
            "orders": {"id", "user_id", "product_id", "amount", "date"}
        }

    @contextmanager
    def get_connection(self):
        """Context manager for safe database connections."""
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
        finally:
            conn.close()

    def query(self, table: str, filters: dict = None) -> list[dict]:
        """
        Query a table with validated table and column names.
        Prevents SQL injection by whitelisting tables and columns.
        """
        # Validate table name
        if table not in self.allowed_tables:
            raise ValueError(f"Table '{table}' not allowed")

        # Build WHERE clause from filters
        where_clause = ""
        params = []
        if filters:
            conditions = []
            for key, value in filters.items():
                if key not in self.allowed_columns[table]:
                    raise ValueError(f"Column '{key}' not allowed in table '{table}'")
                conditions.append(f"{key} = ?")
                params.append(value)
            where_clause = " WHERE " + " AND ".join(conditions)

        # Build and execute query
        query_str = f"SELECT * FROM {table}{where_clause}"
        logger.info(f"Executing query: {query_str}")

        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query_str, params)
            rows = cursor.fetchall()

        # Convert rows to dicts
        columns = [desc[0] for desc in cursor.description]
        return [dict(zip(columns, row)) for row in rows]

# Usage in agent:
db = DatabaseAgent("inventory.db")

def lookup_product(name: str) -> list[dict]:
    """Find products by name."""
    try:
        results = db.query("products", filters={"name": name})
        return results
    except ValueError as e:
        return {"error": f"Query error: {e}"}

This pattern:

• Whitelists tables and columns: Only pre-approved tables/columns can be queried.
• Uses parameterized queries: Values are passed separately, preventing SQL injection.
• Logs every query: Auditable and debuggable.

Monitoring and alerting

Integration health must be monitored:

import json
from datetime import datetime, timedelta

class IntegrationMonitor:
    """Track API health and alert on issues."""

    def __init__(self, alert_threshold_errors: int = 5):
        self.metrics = {}
        self.alert_threshold = alert_threshold_errors

    def log_call(self, service: str, success: bool, latency_ms: float):
        """Record an API call."""
        if service not in self.metrics:
            self.metrics[service] = {
                "total_calls": 0,
                "failed_calls": 0,
                "latencies": [],
                "last_error": None,
                "last_check": None
            }

        m = self.metrics[service]
        m["total_calls"] += 1
        m["last_check"] = datetime.now().isoformat()

        if success:
            m["latencies"].append(latency_ms)
        else:
            m["failed_calls"] += 1
            m["last_error"] = datetime.now().isoformat()

            # Alert if error rate is high
            error_rate = m["failed_calls"] / m["total_calls"]
            if error_rate > 0.1:  # More than 10% errors
                self.alert(f"{service}: High error rate ({error_rate:.1%})")

    def alert(self, message: str):
        """Send an alert (e.g., to Slack, email, logs)."""
        logger.error(f"ALERT: {message}")
        # In production: send_slack_message(message) or similar

    def health_report(self) -> str:
        """Generate a health report."""
        report = []
        for service, m in self.metrics.items():
            if m["total_calls"] == 0:
                continue
            error_rate = m["failed_calls"] / m["total_calls"]
            avg_latency = sum(m["latencies"]) / len(m["latencies"]) if m["latencies"] else 0
            report.append(
                f"{service}: {m['total_calls']} calls, "
                f"{error_rate:.1%} error rate, "
                f"{avg_latency:.0f}ms avg latency"
            )
        return "\n".join(report)

# Usage:
monitor = IntegrationMonitor()

def call_external_api(service: str, request: dict) -> dict:
    """Wrapper that logs metrics."""
    start = time.time()
    try:
        result = api.call(**request)
        latency = (time.time() - start) * 1000
        monitor.log_call(service, success=True, latency_ms=latency)
        return result
    except Exception as e:
        latency = (time.time() - start) * 1000
        monitor.log_call(service, success=False, latency_ms=latency)
        raise

# Periodically report health:
print(monitor.health_report())

Case study: E-commerce fulfillment agent with multi-system integration

Scenario: An e-commerce company built an agent to handle order fulfillment. The agent needed to:

1. Fetch order details from an internal order database (MySQL)
2. Check inventory with a third-party warehouse system (REST API)
3. Request shipment via FedEx API
4. Update order status in the order database
5. Send confirmation email

The system worked in testing but failed in production when:

• Warehouse API had unexpected 10-second latencies
• FedEx API sometimes rejected duplicate requests (idempotency issues)
• Database connections pooled out during peak traffic
• A single failure in step 2 would cascade, leaving inconsistent state

Solution: Implement integration patterns with explicit error handling, idempotency, and circuit breakers:

import uuid
from datetime import datetime, timedelta
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    INVENTORY_CHECKED = "inventory_checked"
    SHIPPED = "shipped"
    FAILED = "failed"

class CircuitBreaker:
    """Prevent cascading failures by stopping requests to a failing service."""

    def __init__(self, service_name: str, failure_threshold: int = 5, timeout_seconds: int = 60):
        self.service_name = service_name
        self.failure_threshold = failure_threshold
        self.timeout = timeout_seconds
        self.failures = 0
        self.last_failure_time = None
        self.state = "closed"  # closed -> open -> half-open -> closed

    def is_open(self) -> bool:
        """Check if circuit is open (rejecting requests)."""
        if self.state == "open":
            elapsed = (datetime.now() - self.last_failure_time).total_seconds()
            if elapsed > self.timeout:
                self.state = "half-open"
                self.failures = 0
                return False
            return True
        return False

    def record_success(self):
        """Record a successful call."""
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        """Record a failed call."""
        self.failures += 1
        self.last_failure_time = datetime.now()
        if self.failures >= self.failure_threshold:
            self.state = "open"
            logger.error(f"Circuit breaker opened for {self.service_name}")

class IdempotentFulfillmentAgent:
    """Handle order fulfillment with idempotency and resilience."""

    def __init__(self, db_pool, warehouse_api, fedex_api):
        self.db = db_pool
        self.warehouse = warehouse_api
        self.fedex = fedex_api

        # Circuit breakers per service
        self.warehouse_cb = CircuitBreaker("warehouse_api", failure_threshold=5)
        self.fedex_cb = CircuitBreaker("fedex_api", failure_threshold=3)

    def fulfill_order(self, order_id: str) -> dict:
        """Execute order fulfillment with full error recovery."""
        idempotency_key = str(uuid.uuid4())

        try:
            # Step 1: Fetch order from database
            order = self.db.query("orders", {"id": order_id})
            if not order:
                return {"error": f"Order {order_id} not found"}

            # Step 2: Check inventory (with circuit breaker)
            if self.warehouse_cb.is_open():
                return {
                    "error": "Warehouse service unavailable; will retry automatically",
                    "order_id": order_id
                }

            try:
                inventory_result = self.warehouse.check_inventory(
                    items=order["items"],
                    timeout=10  # Timeout to prevent hanging
                )
                self.warehouse_cb.record_success()
            except Exception as e:
                self.warehouse_cb.record_failure()
                logger.error(f"Inventory check failed: {e}")
                return {"error": f"Inventory check failed: {e}"}

            if not inventory_result["in_stock"]:
                self.db.update("orders", {"id": order_id, "status": "out_of_stock"})
                return {"status": "out_of_stock", "order_id": order_id}

            # Step 3: Request shipment (with idempotency)
            if self.fedex_cb.is_open():
                return {
                    "error": "Shipping service unavailable; will retry automatically",
                    "order_id": order_id
                }

            try:
                shipment_result = self.fedex.create_shipment(
                    order_details=order,
                    items=inventory_result["allocated_items"],
                    idempotency_key=idempotency_key,  # FedEx API checks this to prevent duplicate shipments
                    timeout=15
                )
                self.fedex_cb.record_success()
            except Exception as e:
                self.fedex_cb.record_failure()
                logger.error(f"Shipment creation failed: {e}")
                return {"error": f"Shipment creation failed: {e}"}

            # Step 4: Update order status (idempotent)
            tracking_number = shipment_result["tracking_number"]
            self.db.update(
                "orders",
                {
                    "id": order_id,
                    "status": "shipped",
                    "tracking_number": tracking_number,
                    "shipped_at": datetime.now().isoformat(),
                    "idempotency_key": idempotency_key  # Store for deduplication
                }
            )

            # Step 5: Send confirmation email (best-effort, don't fail if this fails)
            try:
                self._send_confirmation_email(order, tracking_number)
            except Exception as e:
                logger.warning(f"Email send failed (non-critical): {e}")

            return {
                "status": "success",
                "order_id": order_id,
                "tracking_number": tracking_number
            }

        except Exception as e:
            logger.error(f"Fulfillment failed: {e}")
            self.db.update("orders", {"id": order_id, "status": "failed"})
            return {"error": str(e), "order_id": order_id}

    def _send_confirmation_email(self, order: dict, tracking: str):
        """Send email asynchronously (fire-and-forget)."""
        # Use async queue to prevent blocking
        email_queue.put({
            "customer_email": order["email"],
            "tracking_number": tracking,
            "order_id": order["id"]
        })

Key improvements:

1. Circuit breakers prevent requests to failing services, avoiding cascading failures
2. Idempotency keys ensure FedEx doesn't receive duplicate shipment requests
3. Explicit timeouts on external calls prevent indefinite hangs
4. Best-effort operations (email) don't block the main fulfillment flow
5. Detailed logging helps diagnose issues in production

Result: Order fulfillment reliability improved from 94% to 99.2%. Failed orders were retried automatically without manual intervention.

Edge case: Handling eventual consistency in distributed systems

When integrating with multiple systems, you may face eventual consistency issues. For example:

1. Order database says order is "shipped"
2. Agent calls warehouse to confirm
3. Warehouse API returns order as still "pending" (data hasn't propagated yet)

This creates confusion; the agent might cancel a shipment that's actually already in transit.

Strategy: Temporal consistency checks

def get_order_with_retry(order_id: str, max_retries: int = 3) -> dict:
    """Fetch order status, retrying if sources disagree."""
    for attempt in range(max_retries):
        db_status = db.query("orders", {"id": order_id})["status"]
        api_status = warehouse_api.get_order(order_id)["status"]

        # If they agree, good
        if db_status == api_status:
            return db_status

        # If they disagree, wait and retry (eventual consistency)
        if attempt < max_retries - 1:
            logger.warning(
                f"Status mismatch (DB={db_status}, API={api_status}); "
                f"retrying in 5s..."
            )
            time.sleep(5 + random.random() * 5)  # Backoff
        else:
            # After retries, trust the more authoritative source
            logger.error(f"Consistent mismatch; using API as source of truth")
            return api_status

    return db_status

This pattern handles transient consistency issues without failing the entire order.

Common mistake

Treating external APIs as perfectly reliable and always available. Every external system will fail sometime. Build with this assumption: add retries, timeouts, fallback responses, and monitoring from day one. It's easier to add than to retrofit after the agent crashes in production.