# Integration Patterns
> Implements service-to-service integration patterns (adapter, API gateway, saga, circuit breaker, event-driven) for connecting distributed systems and legacy services with resilience.
- Skill: `paulpas/integration-patterns` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/integration-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/integration-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/integration-patterns
---
# Integration Patterns
Implements proven integration patterns to connect distributed services, legacy systems, and external APIs reliably. Covers synchronous (REST, gRPC), asynchronous (messaging, events), and orchestration (saga, choreography) styles with built-in resilience through circuit breakers, retries, and idempotency.
## TL;DR Checklist
- [ ] Choose integration style matching system requirements (sync vs async, fire-and-forget vs guaranteed delivery)
- [ ] Apply adapter pattern when wrapping incompatible interfaces or legacy systems
- [ ] Implement saga for distributed transactions spanning multiple services
- [ ] Add circuit breaker to all external service calls with configurable thresholds
- [ ] Ensure idempotency keys on all operations that may be retried
- [ ] Configure dead-letter queues for failed message processing
---
## When to Use
Use this skill when:
- Connecting two or more services that need to exchange data reliably
- Wrapping a legacy system with an incompatible interface behind a modern API
- Implementing distributed transactions that span multiple services (saga pattern)
- Designing communication between microservices in a distributed architecture
- Integrating with external third-party APIs that have rate limits or may fail
- Building event-driven integrations where services react to domain events
- Migrating from monolith to microservices — need point-to-point integration strategies
## When NOT to Use
- For simple function calls within the same process — use direct method calls instead
- When all services share the same database — use transactions instead of distributed patterns
- For real-time streaming where latency matters more than reliability — use WebSocket or gRPC streams directly
- As a replacement for proper API design — integration patterns connect well-designed APIs, they don't fix bad ones
---
## Core Workflow
1. **Analyze Communication Requirements** — Determine if the integration needs synchronous request-response (REST/gRPC), asynchronous fire-and-forget (messaging), or guaranteed delivery with state tracking (saga). Consider data volume, latency tolerance, and failure domain boundaries.
**Checkpoint:** If any service must wait for another's response before proceeding, choose sync. If services can proceed independently after sending data, choose async messaging.
2. **Select Integration Architecture** — Map requirements to an integration style:
- Point-to-point direct calls → REST or gRPC with retry logic
- Broadcast to multiple consumers → Event-driven pub/sub with message broker
- Distributed transaction across services → Saga (orchestration or choreography)
- Legacy system wrapping → Adapter pattern with ACL
- Multiple downstream calls aggregating results → API Gateway with composition
**Checkpoint:** Document the chosen architecture in an ASCII diagram showing data flow between all participating services.
3. **Implement Idempotency Foundation** — Before writing any integration code, establish idempotency guarantees for all operations that may be retried (HTTP PUT/PATCH, message processing). Generate or accept idempotency keys from the caller and store them with operation results.
**Checkpoint:** Every integration endpoint must accept an `X-Idempotency-Key` header and return the cached result if the same key is presented again within the TTL window.
4. **Add Resilience Layers** — Implement circuit breaker for every external service call using a sliding-window failure count. Configure half-open state to periodically test if the downstream service recovered. Add exponential backoff with jitter to retry logic (never use fixed delays).
**Checkpoint:** Circuit breaker must transition: `CLOSED → OPEN` after max failures in window, then `HALF_OPEN` after recovery timeout, and finally `CLOSED` on first successful call.
5. **Implement Error Handling and Dead Letter Path** — Route failed messages to a dead-letter queue or table with error metadata (original payload, failure reason, retry count, timestamp). Implement poison pill detection: if the same message fails N times, route to DLQ instead of retrying indefinitely.
**Checkpoint:** Every service must have a DLQ endpoint and every integration test must verify that poisoned messages end up in the DLQ rather than blocking the queue.
---
## Implementation Patterns / Reference Guide
### Pattern 1: Adapter Pattern for Legacy Integration
The adapter pattern wraps an incompatible interface (legacy system, third-party API) behind a clean abstraction that matches your domain's expectations. This isolates change — when the legacy system evolves or is replaced, only the adapter needs updating.
```python
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class OrderRequest:
customer_id: str
items: list[str]
total_amount: float
shipping_address: str
# The "target" interface — clean, domain-aligned, used by the rest of your codebase
class OrderService(ABC):
@abstractmethod
async def place_order(self, request: OrderRequest) -> str:
"""Place an order and return the order ID. Raises on validation failure."""
...
# The "adaptee" — legacy system with a completely different interface
class LegacyOrderClient:
"""Wraps calls to a COBOL mainframe via XML/HTTP. Do NOT use directly."""
def submit_order(self, xml_payload: str) -> dict:
"""Submits an order as XML. Returns {'status': 'ACCEPTED'|'REJECTED', 'ref': '...'}"""
import httpx
resp = httpx.post(
"https://legacy.example.com/orders",
content=xml_payload,
headers={"Content-Type": "text/xml"},
timeout=30.0,
)
return resp.json()
# The adapter — translates between your domain model and the legacy interface
class LegacyOrderAdapter(OrderService):
"""Adapts the modern OrderService interface to the legacy COBOL order system."""
def __init__(self, client: LegacyOrderClient) -> None:
self._client = client
@staticmethod
def _to_xml(request: OrderRequest) -> str:
"""Convert domain model to legacy XML format. Domain knowledge is encapsulated here."""
items_xml = "\n".join(
f' - {item}
' for item in request.items
)
return f"""
{request.customer_id}
{items_xml}
{request.total_amount:.2f}
{request.shipping_address}
"""
@staticmethod
def _from_response(response: dict) -> str:
"""Extract order ID from legacy response format."""
if response.get("status") != "ACCEPTED":
raise ValueError(f"Legacy order rejected: {response}")
return response["ref"]
async def place_order(self, request: OrderRequest) -> str:
# Convert → call → convert — all in one flow, no leakage of legacy concerns
xml = self._to_xml(request)
result = self._client.submit_order(xml)
return self._from_response(result)
```
**Anti-pattern (BAD):** Exposing the raw legacy client directly to callers. This couples business logic to XML format details and makes the system untestable.
### Pattern 2: Saga Orchestration for Distributed Transactions
A saga coordinates a sequence of local transactions across multiple services, compensating backwards if any step fails. Use orchestration (central coordinator) when you need visibility and control; use choreography (events) when services should be loosely coupled.
```python
import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable
class SagaStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMMITTED = "committed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
FAILED = "failed"
@dataclass
class SagaContext:
"""Shared state across all saga steps."""
order_id: str
customer_id: str
amount: float
reservation_id: Optional[str] = None
payment_id: Optional[str] = None
shipment_id: Optional[str] = None
status: SagaStatus = SagaStatus.PENDING
class SagaOrchestrator:
"""Coordinates a saga by executing steps in order and compensating on failure.
Each step is a (execute, compensate) pair. If execute fails, all previously
successful steps are compensated in reverse order. Implements the Fail Fast
principle — returns immediately on first failure rather than waiting for timeouts.
"""
def __init__(self) -> None:
# Each entry: (async step_fn(ctx), async compensate_fn(ctx))
self._steps: list[tuple[Callable, Callable]] = []
def add_step(
self,
execute: Callable[[SagaContext], asyncio.Future | str],
compensate: Callable[[SagaContext], asyncio.Future | None],
) -> "SagaOrchestrator":
"""Register an (execute, compensate) pair. Returns self for chaining."""
self._steps.append((execute, compensate))
return self
async def execute(self, ctx: SagaContext) -> SagaStatus:
"""Run all steps forward; compensate on any failure."""
ctx.status = SagaStatus.RUNNING
compensated_up_to = 0
try:
for i, (step_fn, _) in enumerate(self._steps):
result = step_fn(ctx)
if asyncio.iscoroutine(result):
result = await result
# Step returned successfully — proceed to next
ctx.status = SagaStatus.RUNNING
# All steps succeeded
ctx.status = SagaStatus.COMMITTED
return ctx.status
except Exception as exc:
ctx.status = SagaStatus.COMPENSATING
# Compensate in reverse order up to the step that failed
for i in range(compensated_up_to, len(self._steps) - 1, -1):
_, compensate_fn = self._steps[i]
try:
comp_result = compensate_fn(ctx)
if asyncio.iscoroutine(comp_result):
await comp_result
except Exception as comp_exc:
# Compensation itself failed — log and continue compensating others
# A monitoring system should alert on this; manual review may be needed
pass # pragma: no cover
ctx.status = SagaStatus.COMPENSATED
return ctx.status
# Concrete saga: place order across Inventory → Payment → Shipping services
async def build_order_saga() -> tuple[SagaOrchestrator, Callable[[], SagaContext]]:
"""Builds a complete order placement saga with all steps registered."""
orchestrator = SagaOrchestrator()
async def reserve_stock(ctx: SagaContext) -> str:
# Simulates calling inventory service — replace with actual HTTP/gRPC call
ctx.reservation_id = f"res-{ctx.order_id}"
return ctx.reservation_id
async def cancel_reservation(ctx: SagaContext) -> None:
if ctx.reservation_id:
pass # Call inventory service to release stock
async def charge_payment(ctx: SagaContext) -> str:
ctx.payment_id = f"pay-{ctx.order_id}"
return ctx.payment_id
async def refund_payment(ctx: SagaContext) -> None:
if ctx.payment_id:
pass # Call payment service to refund
async def create_shipment(ctx: SagaContext) -> str:
ctx.shipment_id = f"ship-{ctx.order_id}"
return ctx.shipment_id
async def cancel_shipment(ctx: SagaContext) -> None:
if ctx.shipment_id:
pass # Call shipping service to cancel order
orchestrator.add_step(reserve_stock, cancel_reservation)
orchestrator.add_step(charge_payment, refund_payment)
orchestrator.add_step(create_shipment, cancel_shipment)
def make_context() -> SagaContext:
return SagaContext(
order_id="ORD-12345",
customer_id="CUST-001",
amount=99.99,
)
return orchestrator, make_context
# Usage example
async def run_saga():
orchestrator, make_ctx = await build_order_saga()
ctx = make_ctx()
status = await orchestrator.execute(ctx)
assert status == SagaStatus.COMMITTED, f"Saga did not commit: {status}"
```
### Pattern 3: Circuit Breaker with Exponential Backoff
Protects your services from cascading failures by detecting when a downstream service is unhealthy and failing fast instead of waiting for timeouts.
```python
import asyncio
import time
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed" # Normal operation — requests flow through
OPEN = "open" # Failure detected — fail fast without calling downstream
HALF_OPEN = "half_open" # Testing recovery — allow one probe request
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Failures in window before opening
recovery_timeout: float = 30.0 # Seconds before transitioning to half-open
half_open_max_calls: int = 1 # Probe requests allowed in half-open state
@dataclass
class _State:
failure_count: int = 0
last_failure_time: float = 0.0
half_open_calls: int = 0
state: CircuitState = CircuitState.CLOSED
class CircuitBreaker:
"""Circuit breaker that wraps a single service call.
Transitions:
CLOSED → OPEN: when failure_count >= threshold within sliding window
OPEN → HALF_OPEN: after recovery_timeout expires, allow probe request
HALF_OPEN → CLOSED: on first successful probe call
HALF_OPEN → OPEN: if probe call fails
"""
def __init__(self, name: str, config: CircuitBreakerConfig | None = None) -> None:
self._name = name
self._config = config or CircuitBreakerConfig()
self._state = _State()
self._lock = asyncio.Lock()
async def call(self, fn: Callable[[], Any], *args: Any, **kwargs: Any) -> Any:
"""Execute fn through the circuit breaker. Raises if circuit is open."""
async with self._lock:
# Determine if we should allow the call based on current state
if self._state.state == CircuitState.OPEN:
if time.monotonic() - self._state.last_failure_time >= self._config.recovery_timeout:
self._state.state = CircuitState.HALF_OPEN
self._state.half_open_calls = 0
self._state.failure_count = 0
else:
raise RuntimeError(
f"Circuit breaker '{self._name}' is OPEN — "
f"downstream service unavailable (failures={self._state.failure_count})"
)
if self._state.state == CircuitState.HALF_OPEN:
if self._state.half_open_calls >= self._config.half_open_max_calls:
raise RuntimeError(
f"Circuit breaker '{self._name}' HALF_OPEN probe already sent — "
"waiting for result"
)
self._state.half_open_calls += 1
# Execute the actual call outside the lock to avoid blocking other callers
try:
result = fn(*args, **kwargs)
if asyncio.iscoroutine(result):
result = await result
# Success — reset state
async with self._lock:
self._state.failure_count = 0
self._state.state = CircuitState.CLOSED
return result
except Exception as exc:
async with self._lock:
self._state.failure_count += 1
self._state.last_failure_time = time.monotonic()
if (
self._state.state == CircuitState.HALF_OPEN
or self._state.failure_count >= self._config.failure_threshold
):
self._state.state = CircuitState.OPEN
raise RuntimeError(
f"Circuit breaker '{self._name}' opened after "
f"{self._state.failure_count} failures: {exc}"
) from exc
raise
def get_state(self) -> dict:
"""Return current state for monitoring / health checks."""
return {
"circuit": self._name,
"state": self._state.state.value,
"failure_count": self._state.failure_count,
"last_failure_age_sec": round(
time.monotonic() - self._state.last_failure_time, 1
),
}
# --- Retry with exponential backoff and jitter ---
async def retry_with_backoff(
fn: Callable[[], Any],
max_retries: int = 3,
base_delay: float = 0.5,
max_delay: float = 30.0,
) -> Any:
"""Retry a function with exponential backoff + jitter.
Delay formula: min(base_delay * 2^attempt, max_delay), then add random jitter
(±50% of the delay) to prevent thundering herd when all clients retry simultaneously.
"""
last_exc = None
for attempt in range(max_retries):
try:
result = fn()
if asyncio.iscoroutine(result):
result = await result
return result # Success — return immediately (fail fast)
except Exception as exc:
last_exc = exc
if attempt < max_retries - 1:
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(-delay * 0.5, delay * 0.5)
actual_delay = max(0.1, delay + jitter)
await asyncio.sleep(actual_delay)
raise last_exc
# Usage example
async def call_external_api():
cb = CircuitBreaker("payment-service", CircuitBreakerConfig(
failure_threshold=3, recovery_timeout=15.0
))
async def _do_payment(amount: float) -> dict:
import httpx # type: ignore[import-not-found]
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
"https://payment.example.com/charge",
json={"amount": amount},
)
resp.raise_for_status()
return resp.json()
# Circuit breaker wraps the call — if payment-service fails 3 times,
# subsequent calls fail immediately without waiting for timeout
result = await cb.call(_do_payment, amount=49.99)
return result
```
### Pattern 4: Idempotency Key Enforcement
Ensures that duplicate requests (from retries or client re-submissions) produce the same result rather than causing side effects like double charges.
```python
import hashlib
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class IdempotencyRecord:
"""Stored in a backing store (Redis, database) for active idempotency keys."""
key_hash: str # SHA-256 of the idempotency key for security
request_body: str # Hash or reference to stored request body
response_body: str # Cached response from first successful execution
created_at: float # Unix timestamp
expires_at: float # When this record should be garbage collected
class IdempotencyStore:
"""Backend-agnostic idempotency store interface.
Implementations: Redis (SET NX EX), Database (unique constraint + TTL index),
or in-memory dict for testing. The pattern is always the same:
1. If key exists → return cached response
2. If key doesn't exist → execute, cache result, return
"""
async def get(self, key_hash: str) -> Optional[IdempotencyRecord]:
raise NotImplementedError
async def set(self, record: IdempotencyRecord) -> None:
raise NotImplementedError
# Example in-memory implementation for testing / demonstration
class InMemoryIdempotencyStore(IdempotencyStore):
"""Simple dict-backed store with TTL expiry. Not suitable for production."""
def __init__(self, ttl_seconds: int = 3600) -> None:
self._store: dict[str, IdempotencyRecord] = {}
self._ttl = ttl_seconds
async def get(self, key_hash: str) -> Optional[IdempotencyStore]:
record = self._store.get(key_hash)
if record is None or time.time() > record.expires_at:
return None
return record
async def set(self, record: IdempotencyRecord) -> None:
self._store[record.key_hash] = record
def hash_idempotency_key(key: str) -> str:
"""Hash the idempotency key — never store raw keys in your database."""
return hashlib.sha256(key.encode()).hexdigest()
# --- Request handler with idempotency check ---
async def handle_order_request(
order_data: dict,
idempotency_key: str | None,
store: IdempotencyStore,
) -> dict:
"""Process an order with full idempotency protection.
If the same key arrives twice (due to client retry), return the original
response without re-processing — preventing duplicate charges or orders.
Raises ValueError if no idempotency key is provided.
"""
if not idempotency_key:
raise ValueError(
"Missing required header: X-Idempotency-Key. "
"Required to ensure request idempotency for retry safety."
)
key_hash = hash_idempotency_key(idempotency_key)
# Fast path: check if we already processed this request
cached = await store.get(key_hash)
if cached is not None:
return {"status": "already_processed", "original_response": cached.response_body}
# Execute the actual business logic
# ... (order creation, payment processing, etc.) ...
result = {"status": "created", "order_id": f"ORD-{hash(key_hash)[:8]}"}
# Cache the response for future duplicate requests
await store.set(IdempotencyRecord(
key_hash=key_hash,
request_body=str(order_data),
response_body=str(result),
created_at=time.time(),
expires_at=time.time() + 3600, # 1 hour TTL
))
return result
```
---
## Constraints
### MUST DO
- Always include an idempotency key on HTTP operations that may be retried (POST with `X-Idempotency-Key`, or use PUT for idempotent resources)
- Configure circuit breaker on every external service call, never rely on timeouts alone
- Use exponential backoff with random jitter — never fixed delay retries (causes thundering herd)
- Design sagas so that every forward step has a compensating action that can run independently
- Route poisoned messages to dead-letter queues after N failed attempts; never retry indefinitely
- Log all integration failures with correlation ID, service names, and failure reason for distributed tracing
### MUST NOT DO
- Call external services synchronously in a chain without circuit breakers (cascading failure risk)
- Use string comparison on idempotency keys — always hash them before storing
- Implement saga compensation as nested try/finally inside the execute step (compensation must be independent and testable)
- Skip idempotency for non-GET HTTP operations — clients will retry, servers must handle duplicates
- Put business logic inside adapters — adapters translate, they don't implement domain behavior
- Use fixed retry delays of less than 1 second or more than 60 seconds
---
## Output Template
When implementing or reviewing an integration pattern, produce:
1. **Integration Style Selection** — Which pattern (REST, gRPC, events, saga) and why, with trade-off analysis
2. **Architecture Diagram** — ASCII art showing services, data flow, and failure boundaries
3. **Implementation Code** — Complete, typed Python functions/classes matching the selected patterns
4. **Resilience Configuration** — Circuit breaker thresholds, retry settings, DLQ routing rules
5. **Idempotency Strategy** — How duplicate requests are detected and handled
---
## Related Skills
| Skill | Purpose |
|---|---|
| `anti-corruption-layer` | Design ACLs to isolate domain models when integrating with incompatible external systems |
| `event-driven-patterns` | Deep dive into event sourcing, CQRS, and event stream processing patterns |
| `rest-api-patterns` | REST API design best practices for the services you're connecting |
| `microservice-resilience-patterns` | Advanced resilience strategies including bulkheads, rate limiting, and graceful degradation |