integration-automation-patterns

Reliable enterprise integration patterns: event-driven workflows, system-of-record synchronisation, circuit breaker, saga orchestration, transactional outbox, CDC, and Kafka envelope handling — plus MCP Security Patterns for tool-invocation safety in agentic workflows. See the example catalog below and CI for current test results.

Structural solutions to the recurring failure modes of enterprise integration: duplicate event processing, partial transaction failures, silent data conflicts, and unrecoverable workflow state.

The problem this solves

Enterprise systems that span CRM + ERP + messaging fail in predictable ways. This library provides reference implementations of the patterns that solve these problems structurally — idempotent event envelopes, field-level authority boundaries, saga compensation, transactional outbox, event sourcing with optimistic concurrency, and API gateway composition — so integration logic is explicit, testable, and broker-agnostic.

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

pip install integration-automation-patterns

Optional extras:

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

Quick start

from integration_automation_patterns.event_envelope import EventEnvelope
from integration_automation_patterns.sync_boundary import SyncBoundary, FieldAuthority

# Idempotent event — safe to process multiple times
event = EventEnvelope(
    event_id="evt_8f3a1b",
    source="salesforce",
    event_type="contact.updated",
    payload={"email": "alice@example.com", "phone": "+1-555-0100"},
    schema_version="1.0",
)

# Field-level authority: who owns each field?
boundary = SyncBoundary(
    authorities={
        "email": FieldAuthority(owner="salesforce", read_others=["erp"]),
        "phone": FieldAuthority(owner="erp",        read_others=["salesforce"]),
    }
)
conflicts = boundary.detect_conflict(
    incoming_system="salesforce",
    fields={"email": "alice@example.com", "phone": "+1-555-0100"},
)
# conflicts → {"phone": ConflictDetail(owner="erp", incoming_system="salesforce")}

Example catalog — 43 patterns

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

Workflow and orchestration patterns

Data and messaging patterns

API gateway patterns

Event sourcing and CQRS patterns

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 — Event-driven integration reliability: why idempotency must be structural
• Implementation Note 02 — Idempotency in enterprise event processing
• ADR 001 — 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. Run pytest tests/ -v before opening a pull request.

Citation

@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

License

MIT — see 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.

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.

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.