Agent Integration Patterns

Why This Matters

AI agents don't exist in isolation—they must communicate with databases, external APIs, user interfaces, and other services. Integration patterns define how agents connect to the world. Poor integration leads to fragile systems, data silos, and operational nightmares. Mastering these patterns enables agents that are composable, scalable, and resilient.

Real-World Analogy: Think of integration patterns like a city's transportation network. Roads (REST APIs) handle direct point-to-point travel, postal services (webhooks) deliver notifications asynchronously, and rail systems (event buses) move large volumes of data efficiently. A well-designed city uses all three depending on the need.

Integration Architecture Overview

REST API Client

import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class HTTPMethod(Enum):
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"


@dataclass
class APIConfig:
    base_url: str
    api_key: Optional[str] = None
    timeout: float = 30.0
    max_retries: int = 3
    retry_delay: float = 1.0
    rate_limit_rpm: int = 60
    headers: dict = field(default_factory=dict)


@dataclass
class APIResponse:
    status_code: int
    data: Any
    headers: dict
    latency: float
    cached: bool = False


class AgentRESTClient:
    def __init__(self, config: APIConfig):
        self.config = config
        self.request_count = 0
        self.error_count = 0
        self.cache: dict[str, tuple[Any, float]] = {}
        self.cache_ttl = 300.0
        self.rate_limit_timestamps: list[float] = []

    def _get_headers(self) -> dict:
        headers = {"Content-Type": "application/json", "User-Agent": "AgentSDK/1.0"}
        if self.config.api_key:
            headers["Authorization"] = f"Bearer {self.config.api_key}"
        headers.update(self.config.headers)
        return headers

    def _check_rate_limit(self) -> bool:
        now = time.time()
        self.rate_limit_timestamps = [ts for ts in self.rate_limit_timestamps if now - ts < 60]
        if len(self.rate_limit_timestamps) >= self.config.rate_limit_rpm:
            return False
        self.rate_limit_timestamps.append(now)
        return True

    def _get_cache_key(self, method: HTTPMethod, url: str, params: Optional[dict] = None) -> str:
        key_data = f"{method.value}:{url}:{json.dumps(params or {}, sort_keys=True)}"
        return hashlib.md5(key_data.encode()).hexdigest()

    async def request(
        self, method: HTTPMethod, endpoint: str, data: Any = None,
        params: Optional[dict] = None, use_cache: bool = True,
    ) -> APIResponse:
        if not self._check_rate_limit():
            await asyncio.sleep(1.0)
        if use_cache and method == HTTPMethod.GET:
            cache_key = self._get_cache_key(method, endpoint, params)
            if cache_key in self.cache:
                cached_data, cached_time = self.cache[cache_key]
                if time.time() - cached_time < self.cache_ttl:
                    return APIResponse(status_code=200, data=cached_data, headers={}, latency=0, cached=True)
        url = f"{self.config.base_url}{endpoint}"
        headers = self._get_headers()
        start_time = time.time()
        last_error = None
        for attempt in range(self.config.max_retries):
            try:
                self.request_count += 1
                await asyncio.sleep(0.01)
                response_data = {"status": "success", "data": data}
                latency = time.time() - start_time
                if use_cache and method == HTTPMethod.GET:
                    cache_key = self._get_cache_key(method, endpoint, params)
                    self.cache[cache_key] = (response_data, time.time())
                return APIResponse(status_code=200, data=response_data, headers=headers, latency=latency)
            except Exception as e:
                last_error = e
                self.error_count += 1
                if attempt < self.config.max_retries - 1:
                    await asyncio.sleep(self.config.retry_delay * (attempt + 1))
        raise last_error  # type: ignore

    async def get(self, endpoint: str, params: Optional[dict] = None) -> APIResponse:
        return await self.request(HTTPMethod.GET, endpoint, params=params)

    async def post(self, endpoint: str, data: Any = None) -> APIResponse:
        return await self.request(HTTPMethod.POST, endpoint, data=data, use_cache=False)

    async def put(self, endpoint: str, data: Any = None) -> APIResponse:
        return await self.request(HTTPMethod.PUT, endpoint, data=data, use_cache=False)

    async def delete(self, endpoint: str) -> APIResponse:
        return await self.request(HTTPMethod.DELETE, endpoint, use_cache=False)

    def get_stats(self) -> dict:
        return {
            "total_requests": self.request_count,
            "total_errors": self.error_count,
            "success_rate": (self.request_count - self.error_count) / self.request_count if self.request_count > 0 else 0,
            "cache_size": len(self.cache),
        }

Webhook Handler

import asyncio
import hashlib
import hmac
import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class WebhookEvent(Enum):
    TASK_COMPLETED = "task.completed"
    TASK_FAILED = "task.failed"
    AGENT_ERROR = "agent.error"
    TOOL_RESULT = "tool.result"
    STATE_CHANGED = "state.changed"


@dataclass
class WebhookConfig:
    secret: str
    max_retries: int = 3
    retry_delay: float = 1.0
    timeout: float = 10.0
    signature_header: str = "X-Webhook-Signature"


class WebhookHandler:
    def __init__(self, config: WebhookConfig):
        self.config = config
        self.handlers: dict[WebhookEvent, list[Callable]] = {}
        self.delivery_log: list[dict] = []
        self.failed_deliveries: list[dict] = []

    def register_handler(self, event: WebhookEvent, handler: Callable) -> None:
        if event not in self.handlers:
            self.handlers[event] = []
        self.handlers[event].append(handler)

    def _generate_signature(self, payload: str) -> str:
        return hmac.new(self.config.secret.encode(), payload.encode(), hashlib.sha256).hexdigest()

    def verify_signature(self, payload: str, signature: str) -> bool:
        expected = self._generate_signature(payload)
        return hmac.compare_digest(expected, signature)

    async def receive_webhook(self, payload: dict, signature: str) -> tuple[bool, str]:
        payload_str = json.dumps(payload, sort_keys=True)
        if not self.verify_signature(payload_str, signature):
            return False, "Invalid signature"
        event = WebhookEvent(payload.get("event"))
        handlers = self.handlers.get(event, [])
        if not handlers:
            return True, "No handlers registered"
        for handler in handlers:
            try:
                await asyncio.wait_for(handler(payload.get("data", {})), timeout=self.config.timeout)
            except asyncio.TimeoutError:
                self.failed_deliveries.append({"event": event.value, "timestamp": time.time(), "error": "Handler timeout"})
            except Exception as e:
                self.failed_deliveries.append({"event": event.value, "timestamp": time.time(), "error": str(e)})
        return True, "Delivered"

    async def send_webhook(self, url: str, event: WebhookEvent, data: dict) -> bool:
        payload_str = json.dumps({"event": event.value, "data": data, "timestamp": time.time()}, sort_keys=True)
        signature = self._generate_signature(payload_str)
        for attempt in range(self.config.max_retries):
            try:
                self.delivery_log.append({"url": url, "event": event.value, "timestamp": time.time(), "attempt": attempt + 1, "success": True})
                return True
            except Exception:
                if attempt < self.config.max_retries - 1:
                    await asyncio.sleep(self.config.retry_delay * (attempt + 1))
        self.failed_deliveries.append({"url": url, "event": event.value, "timestamp": time.time(), "error": "Max retries exceeded"})
        return False

    def get_stats(self) -> dict:
        total = len(self.delivery_log) + len(self.failed_deliveries)
        return {
            "total_deliveries": len(self.delivery_log),
            "failed_deliveries": len(self.failed_deliveries),
            "success_rate": len(self.delivery_log) / total if total else 0,
            "handlers_registered": sum(len(h) for h in self.handlers.values()),
        }

Event Bus System

import asyncio
import logging
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class EventType(Enum):
    AGENT_CREATED = "agent.created"
    AGENT_UPDATED = "agent.updated"
    TASK_ASSIGNED = "task.assigned"
    TASK_COMPLETED = "task.completed"
    TASK_FAILED = "task.failed"
    TOOL_REGISTERED = "tool.registered"
    ERROR_OCCURRED = "error.occurred"


@dataclass
class Event:
    id: str
    type: EventType
    data: dict
    timestamp: float
    source: str = ""
    metadata: dict = field(default_factory=dict)


@dataclass
class EventSubscription:
    event_type: EventType
    handler: Callable
    filter_fn: Optional[Callable] = None
    max_retries: int = 3


class EventBus:
    def __init__(self):
        self.subscriptions: dict[EventType, list[EventSubscription]] = {}
        self.event_history: list[Event] = []
        self.max_history_size = 10000
        self.processing_queue: asyncio.Queue = asyncio.Queue()
        self.is_running = False

    def subscribe(self, event_type: EventType, handler: Callable, filter_fn: Optional[Callable] = None) -> None:
        subscription = EventSubscription(event_type=event_type, handler=handler, filter_fn=filter_fn)
        if event_type not in self.subscriptions:
            self.subscriptions[event_type] = []
        self.subscriptions[event_type].append(subscription)

    async def publish(self, event_type: EventType, data: dict, source: str = "") -> str:
        event = Event(id=str(uuid.uuid4()), type=event_type, data=data, timestamp=time.time(), source=source)
        self.event_history.append(event)
        if len(self.event_history) > self.max_history_size:
            self.event_history = self.event_history[-self.max_history_size:]
        await self.processing_queue.put(event)
        return event.id

    async def start_processing(self) -> None:
        self.is_running = True
        while self.is_running:
            try:
                event = await asyncio.wait_for(self.processing_queue.get(), timeout=1.0)
                await self._process_event(event)
            except asyncio.TimeoutError:
                continue
            except Exception:
                continue

    async def _process_event(self, event: Event) -> None:
        subscriptions = self.subscriptions.get(event.type, [])
        for subscription in subscriptions:
            if subscription.filter_fn and not subscription.filter_fn(event):
                continue
            for attempt in range(subscription.max_retries):
                try:
                    await asyncio.wait_for(subscription.handler(event), timeout=10.0)
                    break
                except Exception:
                    if attempt < subscription.max_retries - 1:
                        await asyncio.sleep(0.1 * (attempt + 1))

    def stop_processing(self) -> None:
        self.is_running = False

    def get_event_history(self, event_type: Optional[EventType] = None, limit: int = 100) -> list[Event]:
        filtered = [e for e in self.event_history if e.type == event_type] if event_type else self.event_history
        return filtered[-limit:]

    def get_stats(self) -> dict:
        type_counts: dict[str, int] = {}
        for event in self.event_history:
            type_name = event.type.value
            type_counts[type_name] = type_counts.get(type_name, 0) + 1
        return {
            "total_events": len(self.event_history),
            "event_type_counts": type_counts,
            "subscriptions": sum(len(s) for s in self.subscriptions.values()),
            "queue_size": self.processing_queue.qsize(),
        }

Mathematical Foundations

API Reliability Score:

Webhook Delivery Rate:

Event Processing Latency:

Integration Throughput:

Retry Efficiency:

Performance Considerations

Pattern	Latency	Throughput	Coupling	Best For
REST API	50-500ms	Medium	Tight	Synchronous queries
Webhook	100ms-5s	High	Loose	Event notifications
Event Bus	10-100ms	Very High	Very Loose	Decoupled systems
Message Queue	5-50ms	Very High	Loose	Background processing
GraphQL	100-1000ms	Medium	Medium	Complex data queries
gRPC	5-50ms	Very High	Tight	Internal service calls

Security Considerations

Authentication: Use OAuth2/JWT for user-facing APIs, API keys for service-to-service
Webhook verification: Always verify HMAC signatures on incoming webhooks
Rate limiting: Implement per-client rate limits to prevent abuse
Input validation: Validate and sanitize all incoming data before processing
HTTPS everywhere: Encrypt all API communications in transit
IP whitelisting: Restrict webhook sources to known IP ranges
Audit logging: Log all API calls for security monitoring and compliance
Secret rotation: Regularly rotate API keys and webhook secrets

Interview Questions

1. When should you use REST APIs vs webhooks for agent integration?

Answer: REST APIs are ideal for synchronous request-response patterns where the client needs immediate results. Use for: status checks, simple commands, CRUD operations. Webhooks are ideal for asynchronous notifications where the client doesn't need immediate response. Use for: event notifications, status updates, background task completion. REST is better for real-time needs; webhooks are better for event-driven architectures.

2. How do you handle webhook delivery failures?

Answer: Failure handling: 1) Implement retry with exponential backoff, 2) Use dead letter queues for failed deliveries, 3) Monitor delivery rates and set alerts, 4) Implement idempotency for retries, 5) Provide manual retry capabilities, 6) Log all failures for debugging. Consider: retry limits, timeout handling, and circuit breakers for downstream services.

3. What is the benefit of an event-driven architecture for agents?

Answer: Benefits: 1) Loose coupling between components, 2) Scalability through asynchronous processing, 3) Better fault isolation, 4) Easier to add new consumers, 5) Natural fit for agent workflows. Events enable: pub/sub patterns, event sourcing, CQRS, and reactive systems. Tradeoffs include eventual consistency and increased complexity.

4. How do you secure agent API integrations?

Answer: Security measures: 1) OAuth2/JWT for authentication, 2) API keys for service-to-service, 3) Rate limiting to prevent abuse, 4) Input validation and sanitization, 5) HTTPS everywhere, 6) Webhook signature verification, 7) IP whitelisting for webhooks, 8) Audit logging. Implement defense in depth—never rely on single security control.

5. How do you handle API versioning in agent systems?

Answer: Versioning strategies: 1) URL versioning (/v1/agents), 2) Header versioning (Accept-Version), 3) Query parameter versioning, 4) Content negotiation. Best practices: maintain backward compatibility, deprecate gradually, document changes, provide migration guides. For agents: version tool schemas, response formats, and API contracts separately.

6. What are common integration failure patterns?

Answer: Common failures: 1) Network timeouts, 2) Rate limiting, 3) Authentication failures, 4) Data format mismatches, 5) Service unavailability, 6) Partial responses, 7) Duplicate deliveries. Mitigation: circuit breakers, retry logic, fallback mechanisms, idempotency, and comprehensive monitoring.

7. How do you monitor integration health?

Answer: Monitoring: 1) Track API response times and error rates, 2) Monitor webhook delivery success, 3) Alert on circuit breaker trips, 4) Track event processing latency, 5) Monitor queue depths, 6) Set up synthetic monitoring. Key metrics: latency, throughput, error rates, and availability.

8. How do you test integration patterns?

Answer: Testing approach: 1) Unit tests for individual components, 2) Integration tests for API contracts, 3) Contract testing for API compatibility, 4) Load testing for performance, 5) Chaos testing for resilience, 6) End-to-end tests for complete flows. Use mocks for external services and test failure scenarios explicitly.

Common Pitfalls

Pitfall	Solution
No retry logic	Implement exponential backoff
Ignoring rate limits	Track and respect limits
No idempotency	Design for duplicate handling
Synchronous blocking	Use async for non-critical paths
No circuit breakers	Implement for all external calls
Missing monitoring	Track all integration metrics
Ignoring versioning	Plan for API evolution
No security	Implement authentication everywhere

Summary with Key Takeaways

REST APIs are best for synchronous, request-response patterns
Webhooks enable asynchronous, push-based notifications
Event-driven architectures provide scalability and loose coupling
Circuit breakers prevent cascading failures across integrations
Retry logic with backoff handles transient failures
Monitoring is essential for integration health visibility
Security must be implemented at every integration point
Testing should cover contracts, performance, and failure scenarios

KnowledgeCheck

When should you use REST APIs over webhooks?
- a) For asynchronous notifications
- b) For synchronous request-response
- c) For event-driven systems
- d) For all integrations
What is the purpose of webhook signature verification?
- a) Improve performance
- b) Ensure payload authenticity
- c) Reduce latency
- d) Increase throughput
What is the benefit of event-driven architecture?
- a) Simpler implementation
- b) Loose coupling between components
- c) Synchronous processing
- d) Lower costs
What should webhook retry logic use?
- a) Fixed delay
- b) Exponential backoff
- c) No retries
- d) Random delay
What metric measures API reliability?
- a) Latency
- b) Success rate
- c) Throughput
- d) Cost
Why implement circuit breakers for integrations?
- a) Improve performance
- b) Prevent cascading failures
- c) Reduce costs
- d) Increase accuracy

Answers: 1-b, 2-b, 3-b, 4-b, 5-b, 6-b

Agent Integration Patterns: REST APIs, Webhooks & Event-Driven