Event Driven AI Agents: Python Webhook Guide¶
Polling is a poor fit when your agent should react to a specific change. A mix-review agent does not need to check storage repeatedly. It needs to run when the DAW export finishes.
A webhook makes that flow immediate. But the endpoint must authenticate the sender, enforce an event contract, prevent duplicate runs, queue work, and expose useful execution states. This guide builds that flow with FastAPI, Redis Streams, a worker, polling, and a signed callback sender.
1. Event-Driven AI Agents: Webhooks Instead of Polling¶
An event-driven agent runs when something happens. A polling agent repeatedly asks whether something happened.
Polling consumes compute and API quotas even when there is no work. Its minimum latency is the polling interval. It can also create O(N × M) connections when N agents poll M services. An event-driven design reduces that connection model to O(N + M), according to this overview of event-driven agent patterns.
The basic architecture has three parts:
- A producer emits an event.
- An event bus or queue transports it.
- An agent consumer processes it.
For a music project, your producer might be an export service. After rendering a mix, it emits session.exported. A Python agent consumes that event, reads the project metadata, and prepares review notes.
Treat the event as an immutable, timestamped record of what happened, not an instruction such as review_this_mix_now. A better event says session.exported and includes the relevant session data. This distinction follows the event model described in Atlan’s event-driven architecture overview.
That keeps the producer independent from the agent. You can later attach a loudness-analysis agent or archive agent without changing the export service.
2. Define and Strictly Validate the Python Event Contract¶
A production event schema commonly defines the event name, timestamp, source, payload fields, and schema version. Schema registries can enforce these contracts and stop incompatible changes from breaking downstream agents, according to Atlan’s architecture guide.
Validation must cover the event-specific payload, not only the outer envelope. Create event_model.py with a discriminated union whose event_type selects the payload model:
from datetime import datetime
from typing import Annotated, Literal
from pydantic import (
AnyHttpUrl,
BaseModel,
ConfigDict,
Field,
TypeAdapter,
field_validator,
model_validator,
)
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
class SessionExportedPayload(StrictModel):
project_id: str = Field(min_length=1)
mix_url: str = Field(min_length=1)
class MixReviewRequestedPayload(StrictModel):
project_id: str = Field(min_length=1)
mix_url: str = Field(min_length=1)
requested_by: str = Field(min_length=1)
class BaseEvent(StrictModel):
schema_version: Literal["v1"]
event_id: str = Field(min_length=1)
idempotency_key: str = Field(min_length=1)
source: str = Field(min_length=1)
timestamp: datetime
callback_url: AnyHttpUrl | None = None
@field_validator("timestamp")
@classmethod
def require_timezone(cls, value: datetime) -> datetime:
if value.tzinfo is None:
raise ValueError("timestamp must include a timezone")
return value
@model_validator(mode="after")
def validate_idempotency_key(self):
expected = f"{self.source}:{self.event_type}:{self.event_id}"
if self.idempotency_key != expected:
raise ValueError(f"idempotency_key must equal {expected}")
return self
class SessionExportedEvent(BaseEvent):
event_type: Literal["session.exported"]
payload: SessionExportedPayload
class MixReviewRequestedEvent(BaseEvent):
event_type: Literal["mix.review_requested"]
payload: MixReviewRequestedPayload
EventEnvelope = Annotated[
SessionExportedEvent | MixReviewRequestedEvent,
Field(discriminator="event_type"),
]
event_adapter = TypeAdapter(EventEnvelope)
A valid event looks like this:
{
"schema_version": "v1",
"event_id": "export-a81f",
"idempotency_key": "studio-api:session.exported:export-a81f",
"event_type": "session.exported",
"source": "studio-api",
"timestamp": "2026-07-27T15:42:00Z",
"callback_url": "https://studio.example/webhook-results",
"payload": {
"project_id": "night-drive",
"mix_url": "s3://music-projects/night-drive/mix.wav"
}
}
The models reject missing fields, wrong types, unsupported versions, unknown event types, unexpected fields, and mismatched payloads. They do not verify that a referenced object such as mix_url exists or is authorized for the sender.
Authenticate the raw request before parsing it. The sender uses:
Content-Type: application/json
X-Webhook-Timestamp: 1785166920
X-Webhook-Signature: <lowercase hexadecimal HMAC-SHA256>
The signed bytes and signature are:
ASCII(X-Webhook-Timestamp) + "." + exact raw HTTP body
hex(HMAC-SHA256(WEBHOOK_SECRET, signed byte sequence))
Reformatting the JSON, changing whitespace, or adding a trailing newline changes the signature. The endpoint should read the raw body, verify the timestamp and signature, validate the JSON, authorize the callback destination, and apply idempotency controls.
3. Receive the Webhook, Deduplicate Atomically, and Acknowledge Fast¶
The HTTP request should not wait for model calls or audio analysis. Authenticate it, validate it, claim the idempotency key, enqueue the event, and return.
Install the dependencies:
Create app.py:
import hashlib
import hmac
import os
import time
import uuid
from datetime import datetime, timezone
from urllib.parse import urlparse
import redis
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from event_model import event_adapter
app = FastAPI()
r = redis.Redis.from_url(os.environ["REDIS_URL"], decode_responses=True)
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode()
SIGNATURE_MAX_AGE = int(os.environ["SIGNATURE_MAX_AGE_SECONDS"])
IDEMPOTENCY_TTL = int(os.environ["IDEMPOTENCY_TTL_SECONDS"])
STREAM = os.environ.get("AGENT_STREAM", "agent-events")
CALLBACK_ALLOWED_HOSTS = {
host.strip().lower().rstrip(".")
for host in os.environ.get("CALLBACK_ALLOWED_HOSTS", "").split(",")
if host.strip()
}
CALLBACK_ALLOWED_PORTS = {
int(port)
for port in os.environ.get("CALLBACK_ALLOWED_PORTS", "443").split(",")
if port.strip()
}
if SIGNATURE_MAX_AGE <= 0 or IDEMPOTENCY_TTL <= 0:
raise RuntimeError("Signature age and idempotency TTL must be positive")
CLAIM_AND_ENQUEUE = """
local existing_run = redis.call("GET", KEYS[1])
if existing_run then
local existing_status = redis.call("HGET", KEYS[2] .. existing_run, "status")
return {"duplicate", existing_run, existing_status or "unknown"}
end
local stream_id = redis.call(
"XADD", KEYS[3], "*",
"run_id", ARGV[1],
"event", ARGV[6],
"idempotency_key", ARGV[7]
)
redis.call(
"HSET", KEYS[2] .. ARGV[1],
"status", "accepted",
"event_id", ARGV[3],
"event_type", ARGV[4],
"accepted_at", ARGV[5],
"updated_at", ARGV[5],
"retry_count", "0",
"stream_id", stream_id
)
redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2])
return {"queued", ARGV[1], "accepted"}
"""
RECONCILE_ENTRY = """
local existing_run = redis.call("GET", KEYS[1])
if existing_run and existing_run ~= ARGV[1] then
return "conflict"
end
redis.call("HSETNX", KEYS[2], "status", "accepted")
redis.call("HSETNX", KEYS[2], "event_id", ARGV[2])
redis.call("HSETNX", KEYS[2], "event_type", ARGV[3])
redis.call("HSETNX", KEYS[2], "accepted_at", ARGV[4])
redis.call("HSETNX", KEYS[2], "updated_at", ARGV[4])
redis.call("HSETNX", KEYS[2], "retry_count", "0")
redis.call("HSETNX", KEYS[2], "stream_id", ARGV[5])
if not existing_run then
redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[6])
end
return "reconciled"
"""
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def verify_signature(raw_body: bytes, timestamp: str, signature: str):
try:
sent_at = int(timestamp)
except ValueError as exc:
raise HTTPException(status_code=401, detail="Invalid timestamp") from exc
if abs(int(time.time()) - sent_at) > SIGNATURE_MAX_AGE:
raise HTTPException(status_code=401, detail="Expired request")
signed = timestamp.encode("ascii") + b"." + raw_body
expected = hmac.new(
WEBHOOK_SECRET,
signed,
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(expected, signature):
raise HTTPException(status_code=401, detail="Invalid signature")
def authorize_callback(callback_url):
if callback_url is None:
return
parsed = urlparse(str(callback_url))
hostname = (parsed.hostname or "").lower().rstrip(".")
try:
port = parsed.port or 443
except ValueError as exc:
raise HTTPException(
status_code=403,
detail="Invalid callback port",
) from exc
if (
parsed.scheme != "https"
or parsed.username is not None
or parsed.password is not None
or hostname not in CALLBACK_ALLOWED_HOSTS
or port not in CALLBACK_ALLOWED_PORTS
):
raise HTTPException(
status_code=403,
detail="Callback destination is not approved",
)
def reconcile_boundary_state() -> dict[str, int]:
cursor = "-"
repaired = conflicts = invalid = 0
while True:
entries = r.xrange(STREAM, min=cursor, count=100)
if not entries:
break
for stream_id, message in entries:
cursor = f"({stream_id}"
run_id = message.get("run_id")
raw_event = message.get("event")
key = message.get("idempotency_key")
if not run_id or not raw_event or not key:
invalid += 1
continue
try:
event = event_adapter.validate_json(raw_event)
except ValidationError:
invalid += 1
continue
result = r.eval(
RECONCILE_ENTRY,
2,
f"idempotency:{key}",
f"run:{run_id}",
run_id,
event.event_id,
event.event_type,
now_iso(),
stream_id,
IDEMPOTENCY_TTL,
)
if result == "conflict":
conflicts += 1
else:
repaired += 1
return {
"reconciled": repaired,
"conflicts": conflicts,
"invalid": invalid,
}
@app.post("/webhooks/events")
async def receive_event(
request: Request,
x_webhook_timestamp: str = Header(...),
x_webhook_signature: str = Header(...),
):
raw_body = await request.body()
verify_signature(raw_body, x_webhook_timestamp, x_webhook_signature)
try:
event = event_adapter.validate_json(raw_body)
except ValidationError as exc:
return JSONResponse(
status_code=422,
content={
"status": "rejected",
"error_code": "invalid_event",
"errors": exc.errors(include_url=False),
},
)
authorize_callback(event.callback_url)
run_id = str(uuid.uuid4())
accepted_at = now_iso()
try:
result = r.eval(
CLAIM_AND_ENQUEUE,
3,
f"idempotency:{event.idempotency_key}",
"run:",
STREAM,
run_id,
IDEMPOTENCY_TTL,
event.event_id,
event.event_type,
accepted_at,
event.model_dump_json(),
event.idempotency_key,
)
except redis.RedisError as exc:
raise HTTPException(status_code=503, detail="Queue unavailable") from exc
disposition, stored_run_id, status = result
if disposition == "duplicate":
return {
"status": status,
"run_id": stored_run_id,
"duplicate": True,
"status_url": f"/runs/{stored_run_id}",
}
return JSONResponse(
status_code=202,
content={
"status": "accepted",
"run_id": stored_run_id,
"duplicate": False,
"status_url": f"/runs/{stored_run_id}",
},
)
@app.get("/runs/{run_id}")
def get_run(run_id: str):
record = r.hgetall(f"run:{run_id}")
if not record:
raise HTTPException(status_code=404, detail="Run not found")
return {"run_id": run_id, **record}
Configure and start it:
export REDIS_URL='redis://127.0.0.1:6379/0'
export WEBHOOK_SECRET='replace-with-a-long-random-secret'
export SIGNATURE_MAX_AGE_SECONDS='300'
export IDEMPOTENCY_TTL_SECONDS='604800'
export CALLBACK_ALLOWED_HOSTS='studio.example'
export CALLBACK_ALLOWED_PORTS='443'
uvicorn app:app --host 0.0.0.0 --port 8000
Redis executes a Lua script without another command being interleaved. Two concurrent deliveries therefore cannot both pass the existence check during a successful execution.
That isolation is not transactional rollback. The script calls XADD before creating the run record and claim, so an XADD failure cannot leave a claim for work that was never queued. A later failure can leave a stream entry without boundary state. The reconcile_boundary_state() routine scans existing entries and restores run:{run_id} and the original idempotency claim without calling XADD or creating another execution. Run it at startup and periodically from a maintenance process:
Reconciliation should run before accepting retries after an incident. If a claim already points to a different run, the routine reports a conflict instead of overwriting it.
authorize_callback() requires HTTPS, an explicitly allowed hostname, and an explicitly allowed port. This allowlist is only one SSRF control: DNS rebinding, compromised DNS, redirects, and public hostnames resolving to private addresses remain risks. Disable redirects in the callback client and combine validation with restricted network egress, proxy policy, and destination-IP checks at connection time.
Use a clear response policy:
202: validated and queued.400or422: malformed or invalid event.401or403: failed authentication or authorization.200or409: known duplicate.429: temporary admission limit.500or503: retryable infrastructure failure.
4. Make Idempotency Reliable Under Retries and Concurrency¶
Use a deterministic key:
Do not store it in a Python set. Process-local memory disappears during restarts and is not shared across webhook instances.
Use Redis SET NX, an isolated script, or a database uniqueness constraint. Keep the key for at least the sender’s retry window. Track processing duplicates, completed duplicates, and expired keys separately.
A worker can hold a renewable lease. If it crashes after claiming work, another worker can recover the pending Redis Streams entry. Keep the stable run_id; do not create a second logical run.
Webhook idempotency does not make external effects idempotent. Model calls, storage writes, tool calls, and callback receivers should use run_id or a deterministic operation ID as their downstream idempotency key.
5. Run the Agent and Report a Useful Final Execution Status¶
A 202 response means the webhook was queued, not that the agent completed. Use explicit states:
Create worker.py:
import hashlib
import hmac
import json
import os
import socket
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
import redis
from pydantic import BaseModel, ConfigDict, ValidationError
from event_model import event_adapter
REDIS_URL = os.environ["REDIS_URL"]
STREAM = os.environ.get("AGENT_STREAM", "agent-events")
GROUP = os.environ.get("AGENT_GROUP", "mix-review-workers")
DEAD_LETTER_STREAM = os.environ.get(
"AGENT_DEAD_LETTER_STREAM",
"agent-events-dead-letter",
)
CONSUMER = os.environ.get(
"AGENT_CONSUMER",
f"{socket.gethostname()}-{os.getpid()}",
)
MAX_RETRIES = int(os.environ.get("AGENT_MAX_RETRIES", "3"))
CLAIM_IDLE_MS = int(os.environ.get("AGENT_CLAIM_IDLE_MS", "60000"))
IDEMPOTENCY_TTL = int(os.environ["IDEMPOTENCY_TTL_SECONDS"])
CALLBACK_SECRET = os.environ["CALLBACK_SECRET"].encode()
TERMINAL_STATUSES = {
"succeeded",
"failed_permanent",
"dead_lettered",
}
r = redis.Redis.from_url(REDIS_URL, decode_responses=True)
class MixReview(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
project_id: str
summary: str
review_notes: list[str]
class RetryableAgentError(Exception):
pass
class PermanentAgentError(Exception):
pass
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def ensure_group():
try:
r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
except redis.ResponseError as exc:
if "BUSYGROUP" not in str(exc):
raise
def restore_missing_boundary_state(message: dict[str, str]):
run_id = message["run_id"]
idempotency_key = message.get("idempotency_key")
r.hsetnx(f"run:{run_id}", "status", "accepted")
r.hsetnx(f"run:{run_id}", "retry_count", "0")
r.hsetnx(f"run:{run_id}", "accepted_at", now_iso())
if idempotency_key:
r.set(
f"idempotency:{idempotency_key}",
run_id,
nx=True,
ex=IDEMPOTENCY_TTL,
)
def callback_body(
run_id: str,
status: str,
retry_count: int,
result: dict | None = None,
error_code: str | None = None,
) -> bytes:
document = {
"notification_id": f"{run_id}:{status}:{retry_count}",
"run_id": run_id,
"status": status,
"retry_count": retry_count,
"result": result,
"error_code": error_code,
"occurred_at": now_iso(),
}
return json.dumps(
document,
separators=(",", ":"),
sort_keys=True,
).encode("utf-8")
def send_signed_callback(callback_url: str | None, raw_body: bytes):
if not callback_url:
return
timestamp = str(int(time.time()))
signed = timestamp.encode("ascii") + b"." + raw_body
signature = hmac.new(
CALLBACK_SECRET,
signed,
hashlib.sha256,
).hexdigest()
request = urllib.request.Request(
callback_url,
data=raw_body,
method="POST",
headers={
"Content-Type": "application/json",
"X-Agent-Timestamp": timestamp,
"X-Agent-Signature": signature,
},
)
for attempt in range(3):
try:
with urllib.request.urlopen(request, timeout=10) as response:
if 200 <= response.status < 300:
return
except (urllib.error.URLError, TimeoutError):
time.sleep(2**attempt)
def set_status(
run_id: str,
status: str,
retry_count: int,
callback_url: str | None,
result: dict | None = None,
error_code: str | None = None,
):
record = {
"status": status,
"retry_count": str(retry_count),
"updated_at": now_iso(),
}
if result is not None:
record["result"] = json.dumps(result, separators=(",", ":"))
if error_code is not None:
record["error_code"] = error_code
if status in TERMINAL_STATUSES:
record["completed_at"] = now_iso()
r.hset(f"run:{run_id}", mapping=record)
if status in {
"succeeded",
"failed_retryable",
"failed_permanent",
"dead_lettered",
}:
send_signed_callback(
callback_url,
callback_body(
run_id,
status,
retry_count,
result,
error_code,
),
)
def run_agent(event, run_id: str) -> MixReview:
"""
Pass run_id as the idempotency key to real model, tool, and storage APIs
whenever they support one.
"""
project_id = event.payload.project_id
mix_url = event.payload.mix_url
if not mix_url:
raise PermanentAgentError("missing_mix_url")
return MixReview.model_validate(
{
"project_id": project_id,
"summary": f"Mix review prepared for {project_id}.",
"review_notes": [
"Check vocal balance against the instrumental.",
"Confirm peak and integrated loudness targets.",
],
}
)
def retry_or_dead_letter(message_id, message, event, error_code):
run_id = message["run_id"]
current = int(r.hget(f"run:{run_id}", "retry_count") or "0")
next_retry = current + 1
callback_url = str(event.callback_url) if event.callback_url else None
if next_retry <= MAX_RETRIES:
set_status(
run_id,
"failed_retryable",
next_retry,
callback_url,
error_code=error_code,
)
time.sleep(min(2 ** (next_retry - 1), 30))
r.xadd(
STREAM,
{
"run_id": run_id,
"event": message["event"],
"idempotency_key": message.get("idempotency_key", ""),
"retry_count": str(next_retry),
},
)
r.xack(STREAM, GROUP, message_id)
return
r.xadd(
DEAD_LETTER_STREAM,
{
**message,
"retry_count": str(next_retry),
"error_code": error_code,
"dead_lettered_at": now_iso(),
"source_stream_id": message_id,
},
)
set_status(
run_id,
"dead_lettered",
next_retry,
callback_url,
error_code=error_code,
)
r.xack(STREAM, GROUP, message_id)
def process_message(message_id: str, message: dict[str, str]):
run_id = message.get("run_id")
raw_event = message.get("event")
if not run_id or not raw_event:
r.xadd(
DEAD_LETTER_STREAM,
{
"source_stream_id": message_id,
"error_code": "invalid_stream_message",
"message": json.dumps(message),
"dead_lettered_at": now_iso(),
},
)
r.xack(STREAM, GROUP, message_id)
return
restore_missing_boundary_state(message)
# A crash after persisting a terminal state but before XACK must not
# execute the model or emit the terminal callback again.
if r.hget(f"run:{run_id}", "status") in TERMINAL_STATUSES:
r.xack(STREAM, GROUP, message_id)
return
try:
event = event_adapter.validate_json(raw_event)
except ValidationError:
set_status(
run_id,
"failed_permanent",
int(message.get("retry_count", "0")),
None,
error_code="invalid_queued_event",
)
r.xack(STREAM, GROUP, message_id)
return
callback_url = str(event.callback_url) if event.callback_url else None
retry_count = int(
r.hget(f"run:{run_id}", "retry_count")
or message.get("retry_count", "0")
)
set_status(run_id, "processing", retry_count, callback_url)
try:
review = run_agent(event, run_id)
result = review.model_dump()
except PermanentAgentError as exc:
set_status(
run_id,
"failed_permanent",
retry_count,
callback_url,
error_code=str(exc),
)
r.xack(STREAM, GROUP, message_id)
return
except ValidationError:
set_status(
run_id,
"failed_permanent",
retry_count,
callback_url,
error_code="invalid_agent_output",
)
r.xack(STREAM, GROUP, message_id)
return
except RetryableAgentError as exc:
retry_or_dead_letter(message_id, message, event, str(exc))
return
except Exception as exc:
retry_or_dead_letter(
message_id,
message,
event,
f"unexpected_agent_error:{type(exc).__name__}",
)
return
set_status(
run_id,
"succeeded",
retry_count,
callback_url,
result=result,
)
r.xack(STREAM, GROUP, message_id)
def recover_stale_messages():
response = r.xautoclaim(
STREAM,
GROUP,
CONSUMER,
min_idle_time=CLAIM_IDLE_MS,
start_id="0-0",
count=10,
)
for message_id, message in response[1]:
process_message(message_id, message)
def main():
ensure_group()
while True:
recover_stale_messages()
response = r.xreadgroup(
GROUP,
CONSUMER,
streams={STREAM: ">"},
count=10,
block=5000,
)
for _, messages in response:
for message_id, message in messages:
process_message(message_id, message)
if __name__ == "__main__":
main()
Start the worker separately:
export REDIS_URL='redis://127.0.0.1:6379/0'
export IDEMPOTENCY_TTL_SECONDS='604800'
export CALLBACK_SECRET='replace-with-a-separate-callback-secret'
export AGENT_MAX_RETRIES='3'
python worker.py
The terminal-state check prevents a recovered message from rerunning after succeeded, failed_permanent, or dead_lettered was persisted but before XACK.
Redis status and result writes are keyed by the stable run_id. Callback notifications contain a deterministic notification_id, which the receiver should retain and deduplicate. Real model, tool, and artifact-storage calls must likewise receive run_id as an idempotency key when supported.
This example still uses at-least-once processing. If an external model or storage provider does not support idempotency, a crash after that external effect but before the terminal Redis write can repeat the effect. Similarly, persisting a terminal state before a callback attempt can lose that callback if the process crashes. For stronger guarantees, use an idempotent provider and a transactional callback outbox on its own durable stream. /runs/{run_id} remains the canonical status source.
The callback signature uses:
A receiver must verify the raw body, reject expired X-Agent-Timestamp values, compare X-Agent-Signature in constant time, and deduplicate notification_id.
For the surrounding runtime and artifact model, see the anatomy of an AI job deployment. For multi-step execution, use these Python agent workflow patterns.
6. Test, Observe, and Operate the Webhook in Production¶
Automate tests for validation, signatures, timestamp expiry, replay, simultaneous duplicates, queue outages, boundary reconciliation, worker recovery, model failures, malformed output, and callback delivery.
Run the tests against a disposable Redis database:
export REDIS_URL='redis://127.0.0.1:6379/15'
export WEBHOOK_SECRET='integration-test-secret'
export SIGNATURE_MAX_AGE_SECONDS='300'
export IDEMPOTENCY_TTL_SECONDS='604800'
export CALLBACK_ALLOWED_HOSTS='studio.example'
export CALLBACK_ALLOWED_PORTS='443'
pytest -q
Do not point tests that call flushdb() at a shared or production database.
Log event_id, run_id, event_type, source, and status. Monitor queue age, pending-entry age, execution latency, retries, callback failures, reconciliation conflicts, and dead-lettered events.
Trace the full path from webhook receipt to completion. The same identifiers should appear in the endpoint, worker, model call, storage write, and callback. The Python agent observability guide shows how to connect those signals.
Evolve schemas deliberately, rotate webhook and callback secrets independently, restrict queue permissions, and redact sensitive payloads. Document how to inspect failed runs, reconcile missing boundary state, claim stale entries, quarantine poison messages, and verify that replay did not create duplicate analysis.
Conclusion¶
A reliable event-driven agent starts with a narrow webhook boundary. Verify the timestamp and exact raw body, validate a versioned event-specific payload, authorize callback destinations defensively, isolate concurrent idempotency checks, and reconcile partial Redis writes.
A Redis Streams worker can recover pending work, check terminal states before execution, apply bounded retries, persist explicit statuses, and move exhausted failures to a dead-letter stream. Stable run_id values must also identify downstream model, storage, tool, and callback operations; otherwise at-least-once recovery can repeat external effects.
That pattern works for mix reviews, metadata enrichment, stem checks, and other music workflows triggered by real project changes. Once it behaves correctly under retries and failures, an agent hosting platform can provide the isolated runtime, secrets, persistence, and observability needed to keep it operating.