featuredImage: /assets/integrate-apis-part-5.png

Integrating AI Agents with External APIs - Part 5: Automation Workflows

Parts 1�4 gave the Escalation Concierge reliable surfaces (Slack, Discord, Zapier). Part 5 turns them into a single automation fabric with queueing, retries, and observability so operations teams can trust the agent during real incidents.

Scenario: Unified Incident Orchestrator

When telemetry flags a Sev1 issue, the workflow must:

1. Post a rich Block Kit update in Slack.
2. Notify Discord community moderators.
3. Trigger Zapier to file finance adjustments and Jira tickets.
4. Wait for human acknowledgment inside Slack or Discord.
5. Escalate to PagerDuty if no response within 5 minutes.

We will design this pipeline in layers so each integration remains decoupled but coordinated.

---

Section 1: Workflow Blueprint

Map the moving pieces before you code.

Signal Sources (LLM agent, telemetry)
          |
          v
 Event Router (API Gateway + Queue)
          |
  +-------+--------+
  |                |
Slack Worker   Discord Worker
  |                |
  +-------+--------+
          |
     Zapier Worker
          |
   Finance / CRM

Concern	Pattern	Notes
Transport	Message bus or durable queue	Use Redis Streams, SQS, or Kafka; avoid direct HTTP chaining.
Orchestration	Workflow engine (Temporal, n8n) or code-first queue	Start with queues; migrate to engines when flows exceed 5+ steps.
Idempotency	Dedup store per event_id	Use Redis SET or DynamoDB TTL tables.
Human-in-the-loop	Wait for Slack/Discord reactions	Store pending tasks in DB with SLAs.
Observability	Trace IDs across channels	Attach the same correlation_id to Slack messages, Discord threads, and Zapier tasks.

Document this blueprint in the repo�s /docs/runbooks/automation.md so new engineers can trace every path.

---

Section 2: Event Router and Queueing

Use a thin HTTP service that validates events, enriches them, and drops them on a queue.

Node.js router with BullMQ

// workflows/event-router.js
import express from "express";
import { Queue } from "bullmq";
import crypto from "crypto";

const app = express();
app.use(express.json());

const slackQueue = new Queue("slack-work", { connection: { host: "127.0.0.1", port: 6379 } });
const discordQueue = new Queue("discord-work", { connection: { host: "127.0.0.1", port: 6379 } });
const zapQueue = new Queue("zap-work", { connection: { host: "127.0.0.1", port: 6379 } });

app.post("/events/incidents", async (req, res) => {
  const correlationId = crypto.randomUUID();
  const payload = { ...req.body, correlationId };
  await Promise.all([
    slackQueue.add("incident", payload),
    discordQueue.add("incident", payload),
    zapQueue.add("incident", payload)
  ]);
  res.json({ ok: true, correlationId });
});

app.listen(4010, () => console.log("Event router on 4010"));

Python router with FastAPI + Redis Streams

# workflows/event_router.py
import json
import uuid
import aioredis
from fastapi import FastAPI

app = FastAPI()
redis = aioredis.from_url("redis://localhost:6379", encoding="utf-8", decode_responses=True)

@app.post("/events/incidents")
async def incidents(event: dict):
    correlation_id = str(uuid.uuid4())
    payload = { **event, "correlationId": correlation_id }
    await redis.xadd("incident:stream", payload)
    return {"ok": True, "correlationId": correlation_id}

Router checklist:

1. Validate schema (use Zod, Pydantic) before enqueueing.
2. Reject events without dedupe keys (event_id).
3. Include priority so workers can handle Sev1 before Sev3.

---

Section 3: Workers and Orchestration Logic

Each integration gets its own worker. When flows become more complex, wrap them in Temporal or Prefect, but start with code you can test locally.

Node.js Slack worker

// workflows/workers/slack.js
import { Worker } from "bullmq";
import { postIncidentMessage } from "../../slack/bridge-bot.js";

const worker = new Worker("slack-work", async (job) => {
  const { incidentId, severity, summary, correlationId } = job.data;
  await postIncidentMessage({ incidentId, severity, summary, correlationId });
});

worker.on("completed", (job) => console.log("slack job done", job.id));
worker.on("failed", (job, err) => console.error("slack job failed", job.id, err));

Python Discord worker with Celery

# workflows/workers/discord_worker.py
from celery import Celery
from discord_bridge import post_incident_embed

app = Celery("discord", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")

@app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={"max_retries": 5, "countdown": 15})
def dispatch_incident(self, incident):
    post_incident_embed(
        incident_id=incident["incidentId"],
        severity=incident["severity"],
        summary=incident["summary"],
        correlation_id=incident["correlationId"]
    )

Wait-for-response orchestration

// workflows/waiters/escalation.js
import { getResponseStatus, escalatePagerDuty } from "../services/escalation.js";

export async function waitForAck(correlationId, timeoutMs = 5 * 60 * 1000) {
  const start = Date.now();
  while (Date.now() - start < timeoutMs) {
    const status = await getResponseStatus(correlationId);
    if (status === "acknowledged") return true;
    await new Promise((resolve) => setTimeout(resolve, 5000));
  }
  await escalatePagerDuty(correlationId);
  return false;
}

Guidelines:

1. Keep workers stateless; store workflow state in Redis/Postgres.
2. Retry Slack/Discord/Zapier separately so one failure does not block the others.
3. Use outbox patterns when writing to databases to avoid double sends.

---

Section 4: Reliability Patterns

Idempotency store (Node.js)

// workflows/idempotency.js
import { createClient } from "redis";
const redis = createClient();
await redis.connect();

export async function ensureOnce(eventId, handler) {
  const inserted = await redis.set(`event:${eventId}`, "1", { NX: true, PX: 24 * 60 * 60 * 1000 });
  if (!inserted) {
    console.log("duplicate event", eventId);
    return { skipped: true };
  }
  return handler();
}

Python saga log

# workflows/saga.py
from datetime import datetime

saga_log = {}

def record_step(correlation_id, step, status, metadata=None):
    saga_log.setdefault(correlation_id, []).append({
        "step": step,
        "status": status,
        "metadata": metadata or {},
        "timestamp": datetime.utcnow().isoformat()
    })

def latest_status(correlation_id):
    return saga_log.get(correlation_id, [])

Other safeguards:

1. Dead-letter queues: configure BullMQ limiter and Celery max_retries; push failures to incident:dlq for manual replay.
2. Circuit breakers: pause Zapier worker when Zap failure rate spikes.
3. SLA tracking: store per-step durations to spot bottlenecks (e.g., Slack ack slower than 3 seconds).

---

Section 5: Observability and Runbooks

Trace every workflow through structured logging, metrics, and dashboards shared with incident command.

Node.js OpenTelemetry spans

// observability/workflow-tracing.js
import { context, trace } from "@opentelemetry/api";

const tracer = trace.getTracer("workflow-orchestrator");

export async function tracedStep(name, correlationId, fn) {
  const span = tracer.startSpan(name, { attributes: { correlationId } });
  try {
    return await context.with(trace.setSpan(context.active(), span), fn);
  } catch (err) {
    span.recordException(err);
    span.setStatus({ code: 2, message: err.message });
    throw err;
  } finally {
    span.end();
  }
}

Python Prometheus exporter for queue depth

# observability/queue_metrics.py
import asyncio
from prometheus_client import Gauge, start_http_server
import aioredis

queue_gauge = Gauge("incident_queue_depth", "Messages waiting per queue", ["queue"])

async def monitor():
    conn = aioredis.from_url("redis://localhost:6379", encoding="utf-8", decode_responses=True)
    while True:
        for name in ["slack-work", "discord-work", "zap-work"]:
            size = await conn.llen(f"bull:{name}:wait")
            queue_gauge.labels(queue=name).set(size)
        await asyncio.sleep(5)

if __name__ == "__main__":
    start_http_server(9200)
    asyncio.run(monitor())

Runbook essentials:

1. Replay: xreadgroup from Redis Streams or BullMQ retry command to re-run failed jobs.
2. Pause/Resume: document bullmq.pause("slack-work") and Celery control.cancel_consumer procedures.
3. Token refresh: cross-link to Parts 1�3 for Slack/Discord token recovery.
4. Zap outages: describe fallback when Zapier is down (e.g., queue events and send summary emails later).
5. KPIs: track time-to-ack, queue depth, and percentage of incidents escalated.

---

Implementation Checklist

• Publish the workflow blueprint and correlation-id schema.
• Deploy event routers with schema validation plus dedupe keys.
• Run per-integration workers (Node/Python) with separate retry logic.
• Add idempotency stores, saga logs, and dead-letter queues.
• Wire traces and metrics into your existing observability stack and document replay/runbook steps.

You now have a full-stack integration strategy: disciplined API selection, Slack and Discord surfaces, Zapier automation, and a workflow backbone that keeps everything reliable. Ship it.