Metadata-Version: 2.4
Name: integration-automation-patterns
Version: 0.44.2
Summary: Reference patterns for reliable enterprise integration, workflow automation, and system-of-record synchronization.
Author: Ashutosh Rana
License-Expression: MIT
Project-URL: Homepage, https://github.com/ashutoshrana/integration-automation-patterns
Project-URL: Issues, https://github.com/ashutoshrana/integration-automation-patterns/issues
Project-URL: Changelog, https://github.com/ashutoshrana/integration-automation-patterns/blob/main/CHANGELOG.md
Keywords: enterprise-integration,event-driven,workflow-automation,crm,erp,idempotency,system-of-record,integration-patterns,enterprise-architecture,circuit-breaker,saga-pattern,transactional-outbox,cdc,kafka,webhook
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Financial and Insurance Industry
Classifier: Intended Audience :: Healthcare Industry
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: Office/Business
Classifier: Topic :: Internet :: WWW/HTTP
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: test
Requires-Dist: pytest>=7.0; extra == "test"
Requires-Dist: pytest-cov>=4.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.23; extra == "test"
Provides-Extra: lint
Requires-Dist: ruff>=0.4.0; extra == "lint"
Provides-Extra: typecheck
Requires-Dist: mypy>=1.0; extra == "typecheck"
Provides-Extra: kafka
Requires-Dist: kafka-python>=2.0; extra == "kafka"
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.130.0; extra == "fastapi"
Requires-Dist: httpx>=0.27.0; extra == "fastapi"
Provides-Extra: pydantic
Requires-Dist: pydantic>=2.12.0; extra == "pydantic"
Provides-Extra: mcp
Requires-Dist: mcp<2,>=1.30; extra == "mcp"
Requires-Dist: PyJWT[crypto]<3,>=2.10; extra == "mcp"
Provides-Extra: all
Requires-Dist: integration-automation-patterns[fastapi,kafka,mcp,pydantic]; extra == "all"
Provides-Extra: dev
Requires-Dist: integration-automation-patterns[fastapi,lint,mcp,test,typecheck]; extra == "dev"
Dynamic: license-file

# integration-automation-patterns

[![CI](https://github.com/ashutoshrana/integration-automation-patterns/actions/workflows/ci.yml/badge.svg)](https://github.com/ashutoshrana/integration-automation-patterns/actions/workflows/ci.yml)
[![codecov](https://codecov.io/gh/ashutoshrana/integration-automation-patterns/graph/badge.svg)](https://codecov.io/gh/ashutoshrana/integration-automation-patterns)
[![PyPI](https://img.shields.io/pypi/v/integration-automation-patterns.svg)](https://pypi.org/project/integration-automation-patterns/)
[![Python](https://img.shields.io/pypi/pyversions/integration-automation-patterns.svg)](https://pypi.org/project/integration-automation-patterns/)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
[![Downloads](https://img.shields.io/pypi/dm/integration-automation-patterns.svg)](https://pypi.org/project/integration-automation-patterns/)

Python building blocks and reference examples for integrations that must handle retries, duplicate events, partial failures and conflicting updates. The package includes event envelopes, synchronization boundaries, circuit breakers, sagas and a durable SQLite outbox, plus an optional authenticated MCP resource server.

For backend and platform engineers connecting business systems or building agent tools that enqueue work. You supply the application logic, credentials, broker connections and deployment.

## Start here

| Your task | Start with | Input → result |
|---|---|---|
| Prevent a replayed event from repeating a database change | [Runnable quick start](#quick-start) and [delivery guarantees](#durable-delivery-and-authenticated-mcp) | Event ID + payload + database callback → first processing returns `True`; an identical replay returns `False` |
| Decide which system owns a changed field | [CRM/ERP synchronization example](examples/04_crm_erp_sync_boundary.py) | Field authority rules + incoming updates → conflict information for application handling |
| Recover a workflow after a later step fails | [Saga compensation example](examples/03_saga_compensation.py) | Application steps + compensating callbacks → recorded workflow execution and compensation |
| Let an authenticated agent enqueue work | [MCP setup](#durable-delivery-and-authenticated-mcp) | Verified identity + strict request → durable event or rejection |
| See retrieval, review, execution and handoff together | [Combined demo guide](docs/GOVERNED_SERVICE_DEMO.md) and [executable](examples/governed_service_demo.py) | Synthetic requests → persisted approvals, local business effects, scrubbed handoff and local traces |

The [portfolio guide](https://github.com/ashutoshrana/ashutoshrana/blob/main/PROJECT_GUIDE.md) explains how this package relates to the other libraries.

---

## The problem this solves

Use these patterns when delivery can repeat, workers can restart, or a workflow crosses independent systems. They make retry, ownership and recovery decisions explicit; they do not install a broker, synchronize a CRM automatically or make separate services share one transaction.

For example, an order-ready event may arrive twice after a worker crashes. `SQLiteOutbox.consume` can record the event ID and write the local order effect in the same database transaction, so replay leaves one effect. If the effect is a payment or CRM API call, the remote service needs its own idempotency contract; the local transaction cannot roll that call back.

---

## Architecture overview

```
Inbound Event
     │
     ▼
EventEnvelope (event_id, idempotency_key, source, schema_version)
     │
     ├─ Duplicate Check ──────────► skip if already processed
     │
     ├─ Authority Check ──────────► SyncBoundary.detect_conflict()
     │                               which system owns this field?
     │
     ├─ Retry / Circuit Breaker ──► exponential backoff + OPEN/HALF_OPEN recovery
     │
     └─ Transactional Outbox ─────► write event + domain change in one DB tx
                                     relay delivers to broker independently
```

---

## Installation

```bash
pip install integration-automation-patterns
```

Optional extras:

```bash
pip install 'integration-automation-patterns[kafka]'
pip install 'integration-automation-patterns[fastapi]'
pip install 'integration-automation-patterns[pydantic]'
```

---

## Quick start

This example uses a temporary SQLite database and makes no network calls. Its input is one event delivered twice; its output is one stored business effect.

```python
from tempfile import TemporaryDirectory
from pathlib import Path
from integration_automation_patterns import SQLiteOutbox

with TemporaryDirectory() as directory:
    store = SQLiteOutbox(Path(directory) / "events.sqlite")
    with store.connect() as db:
        db.execute("CREATE TABLE effects (order_id TEXT)")

    def record_order(db, payload):
        db.execute("INSERT INTO effects VALUES (?)", (payload["order_id"],))

    payload = {"order_id": "order-1"}
    print(store.consume("event-1", payload, record_order))  # True
    print(store.consume("event-1", payload, record_order))  # False
    with store.connect() as db:
        print(db.execute("SELECT COUNT(*) FROM effects").fetchone()[0])  # 1
```

Reusing an event ID with a different payload raises `ValueError`. Keep the database file outside a temporary directory to retain deduplication state across application restarts. The callback must use the supplied connection; external API calls are outside this guarantee. See the [outbox example](examples/02_transactional_outbox.py) for the publish/acknowledge boundary and the [combined demo](docs/GOVERNED_SERVICE_DEMO.md) for an actual subprocess crash and replay.

Run the [failure-contract harness](docs/FAILURE_CONTRACT.md) to reproduce process
crashes, lost HTTP acknowledgments, replay with and without downstream idempotency,
confirmed lock contention, a direct-SQL baseline and backlog allocation. It uses
local synthetic services and reports source hashes and measured counts; timings
are local observations, not a production capacity guarantee.


---

## Example catalog — 43 patterns

| # | File | Pattern | Problem Solved |
|---|------|---------|---------------|
| 01 | `01_event_envelope_retry.py` | Idempotent Event | Duplicate processing under at-least-once delivery |
| 02 | `02_transactional_outbox.py` | Transactional Outbox | Reliable event publish in same DB transaction |
| 03 | `03_saga_compensation.py` | Saga | Multi-step distributed transaction with rollback |
| 04 | `04_crm_erp_sync_boundary.py` | Authority Model | Field-level conflict detection across systems |
| 05 | `05_idempotent_crm_update.py` | Idempotent CRM | Deduplication on CRM write operations |
| 06 | `06_end_to_end_integration.py` | End-to-End | Full CRM → ERP integration with all patterns combined |
| 07 | `07_cqrs_event_sourcing.py` | CQRS Basics | Command/query separation with projection |
| 08 | `08_approval_workflow.py` | Approval Workflow | Multi-stage approval with escalation |
| 09 | `09_backpressure_retry.py` | Backpressure | Rate-limited retry with bounded queue |
| 10 | `10_schema_evolution.py` | Schema Evolution | Forward/backward-compatible event schema versioning |
| 11 | `11_process_manager.py` | Process Manager | Long-running process coordination across services |
| 12 | `12_event_streaming_windows.py` | Streaming Windows | Tumbling/sliding window aggregation over event streams |
| 13 | `13_distributed_cache_patterns.py` | Cache Aside | Read-through, write-through, cache invalidation |
| 14 | `14_api_gateway_patterns.py` | API Gateway (v1) | Request routing, header injection, version dispatch |
| 15 | `15_saga_orchestration_patterns.py` | Saga Orchestration | Orchestrator-driven saga with explicit compensation |
| 16 | `16_event_sourcing_cqrs.py` | Event Sourcing (v1) | Append-only event log with basic projection |
| 17 | `17_resilience_patterns.py` | Resilience | Circuit breaker, bulkhead, timeout, retry hierarchy |
| 18 | `18_message_routing_patterns.py` | Message Routing | Content-based router + splitter + aggregator + filter |
| 19 | `19_workflow_orchestration_patterns.py` | Workflow Orchestration | 7-state machine (PENDING→COMPLETED/COMPENSATED) + saga compensation |
| 20 | `20_data_pipeline_patterns.py` | Data Pipeline / ETL | CheckpointStore (fault-tolerant resumability) + DataQualityValidator (5 rule types) + ETLPipeline |
| 21 | `21_api_gateway_patterns.py` | API Gateway (v2) | RateLimiter (token-bucket) + RequestTransformer + ResponseTransformer + APIVersionRouter + APIComposer (sequential + parallel) |
| 22 | `22_event_sourcing_cqrs_patterns.py` | Event Sourcing + CQRS (v2) | AggregateRoot + EventStore (optimistic concurrency) + SnapshotStore + Projection + EventSourcedRepository |
| 23 | `23_service_mesh_resilience.py` | Service Mesh Resilience | Bulkhead (concurrent slot control, BulkheadRejected) + TimeoutHierarchy (operation/service/request tiers) + HealthCheckAggregator (HEALTHY/DEGRADED/UNHEALTHY) + RetryWithJitter (exponential backoff + full jitter) |
| 24 | `24_distributed_tracing_patterns.py` | Distributed Tracing | W3C TraceContext (traceparent header) + Span (OTLP-compatible) + Tracer + AlwaysSample/NeverSample/RatioSampler + InMemorySpanExporter + TracerProvider + W3C Baggage |
| 25 | `25_schema_registry_patterns.py` | Schema Registry | InMemorySchemaRegistry (all 7 CompatibilityLevels) + AvroSchemaEvolution (backward/forward/full) + ConfluentFramingCodec (wire format) + JsonSchemaSerializer + SchemaEvolutionMigrator |
| 26 | `26_graphql_patterns.py` | GraphQL API | GraphQLSchema (SDL builder) + ResolverRegistry (decorator dispatch) + DataLoader (N+1 prevention, batch + prime) + CursorPagination (Relay spec) + FieldAuthorizationPolicy + SubscriptionManager |
| 27 | `27_event_driven_workflow_patterns.py` | Event-Driven Workflows | EventBus (pub/sub + wildcard) + EventSourcingWorkflow (append-only log + state replay) + ChoreographySaga (event-driven saga, no orchestrator) + DeadLetterQueue (retry + DLQ + recovery) + EventFilter (predicate chain + transformer chain) |
| 28 | `28_grpc_streaming_patterns.py` | gRPC Streaming | UnaryUnary (interceptor chain) + ServerStreaming (generator + limit) + ClientStreaming (list + generator drain) + BidirectionalStreaming (None-skip + queue) + StreamingInterceptor (named request/response transforms) |
| 29 | `29_async_concurrent_patterns.py` | Async/Concurrent | AsyncTaskPool (semaphore-bounded + ordered results) + AsyncRetry (exponential backoff + jitter + RetryExhausted) + AsyncPipeline (staged gather + None-drop) + ThreadSafeCache (RLock + TTL) + WorkerPool (ThreadPoolExecutor wrapper) |
| 30 | `30_message_broker_patterns.py` | Message Broker | MessageQueue (priority + FIFO, thread-safe) + TopicExchange (AMQP-style * and # routing) + FanoutExchange (broadcast to all subscribers) + MessageBatch (auto-flush at size) + DeduplicationFilter (time-window idempotency) |
| 31 | `31_service_discovery_patterns.py` | Service Discovery | ServiceRegistry (register/deregister/heartbeat/evict_stale, thread-safe) + LoadBalancer (round_robin/random/weighted strategies) + HealthChecker (configurable failure_rate simulation) + ServiceMesh (route/rebalance/register_and_check orchestration) |
| 32 | `32_observability_patterns.py` | Observability | MetricRegistry (thread-safe counters/gauges, snapshot, reset) + Histogram (configurable bucket boundaries, cumulative counts, percentile via linear interpolation) + SpanTracer (uuid-based trace/span recording, duration_ms, get_trace, completed_spans) + HealthCheck (check registry, run_all, is_healthy, summary) + AlertRule/AlertManager (threshold-based alerting, 5 condition types, fired_count accumulation) |
| 33 | `33_data_pipeline_patterns.py` | Data Pipelines | BatchProcessor (fixed-size batching + retry with per-batch retry loop + thread-safe stats) + StreamProcessor (tumbling/sliding windows, process_stream, aggregate) + ETLPipeline (fluent builder: extract/transform/load, run/run_dry/reset, timing) + DataLineage (directed graph, BFS lineage_path, upstream/downstream, to_dict) + DataQualityChecker (named rule registry, check_batch with violation counts, valid_only) |
| 34 | `34_distributed_transaction_patterns.py` | Distributed Transactions | TwoPhaseCommit (2PC coordinator: prepare/commit/rollback with participant vote) + CompensatingTransaction (Saga-style: LIFO compensation on failure, method-chaining add_step) + DistributedLock (token-based TTL lock, reentrant support, TTL extension, blocking acquire with timeout) + TransactionCoordinator (multi-resource coordinator: LIFO rollback, UUID-keyed enlisted resources) |
| 35 | `35_event_sourcing_advanced.py` | Event Sourcing Advanced | EventProjector (handler dispatch + state accumulation, reset) + EventUpcaster (version-ordered schema migrations, current_version) + EventReplay (since/by_aggregate/by_type/between filtering) + CausationChain (causation_id + correlation_id tracking, chain walk with cycle guard, correlation_group) + TemporalQuery (state_at point-in-time replay, events_in_window, latest_before) |
| 36 | `36_workflow_state_machine.py` | Workflow State Machine | StateMachine (FSM: add_state on_enter/on_exit + add_transition guard+action + trigger + can_trigger) + HierarchicalStateMachine (composite states: add_child_state + set_initial_child + enter_state auto-descend + is_in_state ancestor walk) + WorkflowEngine (multi-step orchestrator: per-step max_retries + exponential backoff + timeout_ms via ThreadPoolExecutor + WorkflowResult) + ConditionalRouter (first-match routing: register_route + register_default + route, ValueError on no match) + ParallelWorkflow (ThreadPoolExecutor: run with timeout + run_until_first_success cancels remaining) |
| 37 | `37_rate_limiting_patterns.py` | Rate Limiting / Throttling | SlidingWindowRateLimiter (deque-based per-client: allow/remaining/reset_at, thread-safe) + TokenBucketLimiter (burst capacity: lazy refill via monotonic(), consume(tokens)/available_tokens) + AdaptiveThrottler (AIMD: record_success/record_failure×0.5, allow() via internal token bucket) + QuotaManager (3-tier per-second/per-minute/per-day: atomic all-or-nothing consume, status, reset) + CircuitBreakerRateLimiter (sliding window + CLOSED/OPEN/HALF_OPEN: call(fn) raises RateLimitExceeded or CircuitOpen) |
| 38 | `38_reactive_streams_patterns.py` | Reactive Streams / Backpressure | BatchProcessor (configurable batch_size + flush_timeout_ms: thread-safe accumulate/flush, stats) + StreamProcessor (generator pipeline: transform/filter/window stage chaining, drain) + ETLPipeline (extract→transform→load with per-record error handling and ETLResult stats) + DataLineage (node/edge DAG: add_node/add_edge, get_upstream/get_downstream DFS, impact_analysis) + DataQualityChecker (completeness/uniqueness/range/pattern/custom rules, quality_score 0–1, DataQualityError) |
| 39 | `39_distributed_tracing_patterns.py` | Distributed Tracing / Observability | TraceContext (W3C traceparent: uuid4 trace_id/span_id, to_traceparent/from_traceparent, child_context(), sampling_flag) + Span (start/finish timing, add_event, set_status OK/ERROR/UNSET, to_dict() serialization, live duration_ms) + Tracer (active+completed span lists, threading.Lock, get_trace sorted by start_time, export) + SamplingStrategy (deterministic hash-based: always_on/always_off/ratio, should_sample) + TraceExporter (get_by_service/get_errors/get_slow_spans filters, clear) |
| 40 | `40_saga_choreography_patterns.py` | Saga Choreography / Distributed Transactions | SagaEvent (frozen dataclass: event_id/saga_id/event_type/payload/timestamp/sequence, create() classmethod UUID factory) + SagaParticipant (execute/compensate with event recording, executed_count/compensated_count) + SagaChoreographer (event bus: register_handler, start/complete/fail_saga, publish_event collects responses, get_saga_status/events) + CompensationEngine (reverse-order rollback: record_execution, compensate_all() reverses execution list, compensation_count) + SagaMonitor (configurable timeout_seconds, record_start/end, is_timed_out, get_stats 5-bucket, get_timed_out_sagas) |
| 41 | `41_api_versioning_patterns.py` | API Versioning / Contract Testing | SemanticVersion (frozen dataclass: parse "1.2.3", is_compatible_with same major + minor>=, is_breaking_change_from major>, rich comparisons) + APIContract (validate_request missing required fields, validate_response expected fields, is_backward_compatible_with older contract) + VersionRouter (exact-match first then highest-compatible-minor fallback, ValueError when no match, available_versions/latest_version) + ChangelogEntry (BREAKING/FEATURE/FIX, is_breaking()) + APIVersionManager (register_version, deprecate, get_supported_versions non-deprecated, get_changelog/get_breaking_changes with optional since_version filter) |

**Agentic & Runtime Security (2025–2026 Standards)**

| 42 | `42_mcp_security_patterns.py` | MCP / Tool Security | MCPToolDefinition (mutable illustrative manifest with canonical SHA-256 checksum), MCPSecurityValidator (source allowlist + metadata + permission + checksum integrity), MCPToolRegistry (origin-allowlisted registration + duplicate detection), MCPInvocationGuard (pre-invocation policy enforcement + full audit trail), MCPRateLimiter (per-tool sliding-window rate limit) — illustrative local checks; not an authenticated MCP transport or a guarantee against prompt injection |
| 43 | `43_agentic_security_auditor.py` | Enterprise Agentic Security Audit | Holistic agentic AI security gap-analysis framework — AgenticSystemConfig (25-field configuration across 7 domains), AgenticSecurityAuditor (28 controls: Tool Permission, MCP Security, Identity & Auth, Memory & Context, Multi-Agent, Observability, HITL; conditional SKIP for undeployed domains), AgentAuditReport (scored 0–100, Sandbox/Controlled/Trusted/Autonomous maturity), framework refs: OWASP ASI 2026, OWASP LLM 2025, NIST AI 600-1, MITRE ATLAS v5.1, CSA ATF, CVE-2025-6514, SOC 2, ISO 42001 |

---

## Pattern reference

### Core integration patterns

| Pattern | Class | Description |
|---------|-------|-------------|
| Idempotent Event | `EventEnvelope` | UUID event_id deduplication under at-least-once delivery |
| Authority Model | `SyncBoundary` | Field-level conflict detection — which system of record owns this field? |
| Circuit Breaker | `CircuitBreaker` | CLOSED → OPEN → HALF_OPEN state machine with configurable thresholds |
| Bulkhead | `Bulkhead` | Max concurrent + queue control; `BulkheadRejected` when both full |
| Timeout Hierarchy | `TimeoutHierarchy` | Three-tier: operation → service → request; `TimeoutExceeded(tier, budget_ms)` |
| Health Check Aggregator | `HealthCheckAggregator` | HEALTHY/DEGRADED/UNHEALTHY; `is_ready()` returns False only when UNHEALTHY |
| Retry with Jitter | `RetryWithJitter` | Exponential backoff + full jitter; `RetryExhausted(attempts, last_error)` |
| Saga | `SagaOrchestrator` | Multi-step distributed transaction with reverse-order compensation |
| Transactional Outbox | `OutboxPublisher` | Reliable publish in same DB transaction as domain change |
| CDC Event | `CDCEvent` | Typed INSERT/UPDATE/DELETE with Debezium-compatible parsing |
| Webhook Validation | `WebhookHandler` | HMAC-SHA256 signature verification + idempotency |

### Workflow and orchestration patterns

| Pattern | Class | Description |
|---------|-------|-------------|
| Workflow State Machine | `WorkflowStateMachine` | 7-state lifecycle: PENDING → RUNNING → COMPLETED / FAILED / COMPENSATING / COMPENSATED / CANCELLED |
| Compensation Engine | `CompensationEngine` | Reverse-order saga rollback with per-step compensating activities |
| Process Manager | `ProcessManager` | Long-running correlation across multiple services and events |
| Approval Workflow | `ApprovalWorkflow` | Multi-stage approval with timeout and escalation |

### Data and messaging patterns

| Pattern | Class | Description |
|---------|-------|-------------|
| Data Quality Validator | `DataQualityValidator` | COMPLETENESS / UNIQUENESS / RANGE / REGEX / CUSTOM rules with `QualityGateError` |
| Checkpoint Store | `CheckpointStore` | Stage + record-level resumability for fault-tolerant ETL |
| Content-Based Router | `ContentBasedRouter` | Route messages to handlers by payload predicate |
| Message Aggregator | `MessageAggregator` | Collect N messages into a batch with timeout |
| Streaming Windows | `TumblingWindow`, `SlidingWindow` | Event stream aggregation |

### API gateway patterns

| Pattern | Class | Description |
|---------|-------|-------------|
| Rate Limiter | `RateLimiter` | Token-bucket per client, thread-safe, raises `RateLimitExceeded(retry_after_seconds)` |
| Request Transformer | `RequestTransformer` | Header injection, body field rename, URL prefix rewrite |
| Response Transformer | `ResponseTransformer` | Status code remapping, response field masking (`"***"`) |
| Version Router | `APIVersionRouter` | Handler registry by version string; raises `UnsupportedVersion` |
| API Composer | `APIComposer` | Fan-out to multiple backends; sequential and parallel merge |

### Event sourcing and CQRS patterns

| Pattern | Class | Description |
|---------|-------|-------------|
| Aggregate Root | `AggregateRoot` | `on_<EventType>` handler dispatch, uncommitted event collection |
| Event Store | `EventStore` | Append-only with optimistic concurrency; `ConcurrencyConflict` on version mismatch |
| Snapshot Store | `SnapshotStore` | Aggregate state checkpoint to reduce event replay time |
| Projection | `AccountBalanceProjection` | Read model built from event stream; `rebuild()` and `reset()` |
| Repository | `EventSourcedRepository` | EventStore + SnapshotStore with auto-snapshot at configurable threshold |

---

## Repository structure

```
src/integration_automation_patterns/
├── event_envelope.py         # Reliable event transport + retry + delivery status
├── sync_boundary.py          # Bi-directional SOR sync with field authority
├── circuit_breaker.py        # CLOSED/OPEN/HALF_OPEN circuit breaker
├── saga.py                   # Distributed saga orchestrator + compensation
├── outbox.py                 # Transactional outbox for at-least-once delivery
├── kafka_envelope.py         # Kafka envelope: partition key, DLQ, schema version
├── webhook_handler.py        # HMAC-SHA256 webhook verification + idempotency
└── cdc_event.py              # CDC event types (Debezium-compatible)
examples/                     # 30 runnable pattern examples (see catalog above)
tests/                        # 977 passing tests
docs/
├── architecture.md
├── implementation-note-01.md # Event-driven integration reliability
├── implementation-note-02.md # Idempotency in enterprise event processing
└── adr/
    └── 001-idempotency-key-design.md
```

---

## Published notes

- [Implementation Note 01](./docs/implementation-note-01.md) — Event-driven integration reliability: why idempotency must be structural
- [Implementation Note 02](./docs/implementation-note-02.md) — Idempotency in enterprise event processing
- [ADR 001](./docs/adr/001-idempotency-key-design.md) — Idempotency key design: event_id as the deduplication unit

---

## Near-term roadmap

- GraphQL subscriptions with WebSocket transport
- Kafka + FastAPI integration examples with async support
- gRPC bidirectional streaming patterns

---

## Contributing

Read [CONTRIBUTING.md](./CONTRIBUTING.md). Run `pytest tests/ -v` before opening a pull request.

---

## Citation

```bibtex
@software{rana2026iap,
  author  = {Rana, Ashutosh},
  title   = {integration-automation-patterns: Enterprise integration reliability patterns},
  year    = {2026},
  version = {0.43.0},
  url     = {https://github.com/ashutoshrana/integration-automation-patterns},
  license = {MIT}
}
```

---

## Part of the enterprise AI patterns trilogy

| Library | Focus | Coverage |
|---------|-------|---------|
| [enterprise-rag-patterns](https://github.com/ashutoshrana/enterprise-rag-patterns) | What to retrieve | 50 sectors · 65+ regulations · 25 jurisdictions · 9 AI frameworks · 1,901 tests |
| [regulated-ai-governance](https://github.com/ashutoshrana/regulated-ai-governance) | What agents may do | 41 governance examples · 25 jurisdictions · 2,631 tests |
| **integration-automation-patterns** | How data flows | 43 patterns · schema registry · GraphQL · 1,865 tests |

---

## License

MIT — see [LICENSE](LICENSE).


## Durable delivery and authenticated MCP

`SQLiteOutbox` is a file-backed reference for **at-least-once delivery**. Write the
business change and event through `enqueue(..., business_write=callback)` in one
transaction. Relay `pending()` records, publish, then `mark_published(id)`. A crash
between those last two operations can cause duplicate delivery. The consumer's
`consume(id, payload, effect)` commits the inbox ID and the callback's database
writes together; the callback must use the supplied connection. External API calls
are not part of that transaction and still require downstream idempotency keys.
SQLite serializes writers; this reference is intended for modest throughput.

```python
from integration_automation_patterns import SQLiteOutbox

store = SQLiteOutbox("events.sqlite")
store.enqueue("event-1", {"status": "ready"})
for event_id, payload in store.pending():
    broker_publish(event_id, payload)  # application-owned broker implementation
    store.mark_published(event_id)
# Consumer: effect(db, payload) must write using db, not an external API.
store.consume(event_id, payload, effect)
```

Install `pip install 'integration-automation-patterns[mcp]'` for the real MCP SDK
resource-server integration. The SDK v1 line is explicitly bounded (`>=1.30,<2`);
this is not a v2 compatibility claim. Validated locally with MCP 1.30.0, PyJWT
2.14.0 and Pydantic 2.13.5. Core users need no MCP dependencies.

```python
from integration_automation_patterns.mcp_server import JWTVerifier, create_mcp_server

verifier = JWTVerifier(
    public_key=trusted_public_key_pem,  # operator-configured issuer public key
    issuer="https://identity.example",
    audience="https://events.example/mcp",
    grants={"authorized-user": frozenset({"events:write"})},
)
server = create_mcp_server(
    store, verifier, audit_sink=write_redacted_audit,
    approved_permissions=frozenset({"events:write"}),
)
app = server.streamable_http_app()  # serve with app lifespan and HTTPS termination
```

The adapter targets MCP SDK v1 (`>=1.30,<2`). Automated major-version updates
are excluded because v2 removes v1 imports used by this adapter; v1 updates
remain enabled. A v2 migration must port the adapter and pass the HTTP auth,
strict-schema, replay and audit tests before widening the dependency range.

The audience URL also configures the HTTP endpoint path (including nested paths).
Its HTTPS origin configures browser-origin validation; credentials, query strings
and fragments are rejected in resource URLs.

The server uses SDK authentication middleware and RS256 signature, issuer,
audience and expiry verification. Effective scopes intersect token scopes with
operator-configured user grants. The tool accepts only
`{"request": {"request_id": "id-1", "event_type": "created", "value": "text"}}`.
It rejects extra fields and non-string/nested values. IDs are scoped per verified
subject; replay returns `created: false`, while reusing an ID with different data
fails. Manifest permissions must equal operator approval at startup. The audit
callback records only the tool and decision; no token, subject or argument values.
The supplied callback must be durable if your deployment requires durable audits;
its acknowledgment failure before enqueue blocks execution. Async sinks and
awaitable sink results are rejected. If the post-enqueue acknowledgment fails,
`MCPAuditDeliveryError(event_committed=True)` reports that the event is already
durable (the SDK includes this marker in the tool error). Retry only with the same
request ID and payload to preserve deduplication; `event_committed=False` means
this call did not enqueue. Pre-authentication rejections are returned by the
SDK's HTTP layer. Configure issuer key rotation/revocation and HTTPS at deployment.

`examples/42_mcp_security_patterns.py` remains an educational manifest-checking
example, separate from this authenticated server. Its digest is not a signature;
trusted expected digests must be supplied independently.
