Coordinating AI Agents With AWS SQS: A Practical Queue-Based Architecture

Building a single AI agent is not usually the hard part.

You send a prompt to a model, get a response back, and wire it into your app. Done.

The hard part starts when that agent becomes one step in a larger system. A real AI workflow might need to ingest a file, extract text, chunk it, generate embeddings, call an LLM, write results to a database, sync to an external API, and notify a user. Those steps do not behave the same. Text extraction might finish in seconds. An LLM call might take minutes. A sync job might fail because some external API is having a bad day. That is where a lot of "agent" systems stop looking magical and start looking like regular distributed systems.

I have seen this fail in boring ways:

• The same job gets processed twice.
• A worker writes to the database, then crashes before marking the job complete.
• A model call runs longer than expected and the message gets picked up again.
• A retried tool call creates duplicate external writes.
• Failed jobs sit in processing until someone manually checks the database.

None of this is new. AI agents do not magically avoid old infrastructure problems. They still need queues, retries, idempotency, durable state, and monitoring.

AWS SQS is a good fit for that middle layer. It is not a full workflow engine. I would not use it for every orchestration problem. But if you need a durable queue between independent agent stages, SQS is simple, reliable, and usually enough.

The Coordination Problem

A basic multi-stage AI workflow often looks like this:

Input source -> ingestion -> processing -> generation -> sync

The first version is usually a database table with a status column. That works for a while.

Then concurrency shows up.

Two workers read the same pending row. A process crashes and leaves a job stuck in processing. Someone adds sleep(30) because the previous step "usually finishes by then."

That last one is the kind of fix that works just long enough to become a production bug.

A queue gives each stage a cleaner boundary. One stage publishes work. Another stage consumes it. If the next stage slows down, the queue absorbs the backlog instead of forcing the whole pipeline to wait.

Input Source
 -> ingest_queue -> Ingestion Worker
 -> chunk_queue -> Chunking Worker
 -> embedding_queue -> Embedding Worker
 -> summary_queue -> Summary Worker
 -> sync_queue -> Sync Worker

Now ingestion can scale separately from summarization. If LLM generation is slow, messages pile up in summary_queue. That is not automatically a failure. That is what the queue is there for.

A failed summary worker does not corrupt the whole workflow. The message can be retried. If it keeps failing, it moves to a dead letter queue.

Standard Queues vs. FIFO Queues

SQS gives you two main queue types: standard queues and FIFO queues.

Standard Queues

Standard queues give at-least-once delivery and best-effort ordering.

A message can be delivered more than once. Messages may not arrive in the exact order sent. That sounds scary, but most background AI work should already handle this.

Use standard queues for work like document processing, embedding generation, batch classification, independent user requests, and webhook processing. For these jobs, throughput matters more than strict ordering.

FIFO Queues

FIFO queues preserve ordering within a MessageGroupId and support deduplication. Use when sequence actually matters: conversation turns, per-user workflows, ordered state transitions.

response = sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps(payload),
    MessageGroupId=payload["user_id"],
    MessageDeduplicationId=payload["task_id"]
)

Be careful with the group ID. If every message uses the same MessageGroupId, you have serialized the whole queue by accident. Give each conversation, user, or workflow its own group ID so you preserve ordering per entity while allowing parallelism across different ones.

My default rule: start with standard queues unless ordering is clearly required. Then make the handler idempotent. That matters more than the queue type.

Ensuring Idempotency in Your Agent Flow

Idempotency means the same task can run more than once without creating duplicate or incorrect side effects. This is the part I would not skip.

SQS standard queues use at-least-once delivery, so duplicates are part of the contract. But this matters even more with AI workloads because model calls are expensive and outputs can be non-deterministic. Retrying the same prompt may cost money and return a different answer. Retrying the same tool call may send a duplicate email or write a second database row.

The basic pseudo workflow:

receive message
check if task already completed
if completed, delete message and exit
if not completed, process task
store result
delete message

Simple version:

def handle_message(message, store, sqs, queue_url):
    payload = json.loads(message["Body"])
    task_id = payload["task_id"]

    if store.already_completed(task_id):
        sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"])
        return {"status": "skipped", "task_id": task_id}

    result = run_agent_logic(payload)
    store.mark_completed(task_id, result)
    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"])
    return {"status": "completed", "task_id": task_id}

The store can be Postgres, DynamoDB, Redis, or anything durable with atomic writes. For Postgres, a unique constraint saves you:

CREATE TABLE agent_task_results (
    task_id TEXT PRIMARY KEY,
    status TEXT NOT NULL,
    result JSONB
);

INSERT INTO agent_task_results (task_id, status)
VALUES ($1, 'processing')
ON CONFLICT (task_id) DO NOTHING;

If the insert succeeds, this worker owns the task. If it does nothing, another worker already claimed or completed it.

The Failure Case I Designed Around

summary_queue -> Summary Worker -> Postgres -> sync_queue

The summary worker receives a message, calls an LLM, writes the summary to Postgres, then deletes the SQS message.

Now suppose the worker writes to Postgres but crashes before deleting the SQS message.

From SQS's point of view, the job never finished. After the visibility timeout expires, another worker receives the same message and runs the task again.

Without idempotency, that retry may call the LLM again, generate a slightly different summary, and write a second result.

A safer handler checks whether model output already exists before calling the model:

def summary_handler(payload, store):
    task_id = payload["task_id"]
    existing = store.get(task_id)

    if existing and existing.get("model_output"):
        summary = existing["model_output"]
    else:
        text = load_text(payload["input"]["text_uri"])
        summary = call_llm(text)
        store.save_model_output(task_id, summary)

    store.save_final_result(task_id, {"summary": summary})
    return {"next_stage": "sync", "next_input": {"summary_task_id": task_id}}

That avoids repeating the expensive part if the first attempt already got that far.

Visibility Timeout

When a worker receives a message, SQS hides it from other workers for the visibility timeout. If the worker finishes, it deletes the message. If the worker crashes, the message becomes visible again after the timeout expires.

Too short: another worker receives the same message while the first is still running. Duplicate execution.

Too long: failed jobs take too long to retry.

visibility_timeout = 2x to 5x expected processing time

Reference:

• Metadata validation: 30-60 seconds
• Embedding generation: 1-5 minutes
• LLM-heavy summary: 5-15 minutes
• Long document analysis: 15+ minutes with heartbeat

For long-running tasks, extend visibility:

sqs.change_message_visibility(
    QueueUrl=queue_url,
    ReceiptHandle=receipt_handle,
    VisibilityTimeout=extension_seconds
)

The message should describe the work, not carry the workload.

Bad:

{"task_id": "123", "full_pdf_text": "... thousands of lines ..."}

Better:

{
  "task_id": "123",
  "stage": "summarize",
  "input": {"document_uri": "s3://bucket/docs/input.pdf"},
  "metadata": {"user_id": "789", "priority": "normal"}
}

Store large files in S3. Send references through SQS. Do not let the queue become your storage layer.

Dead Letter Queues

A DLQ captures messages that fail repeatedly. Without one, poison messages cycle forever.

sqs.set_queue_attributes(
    QueueUrl=main_queue_url,
    Attributes={
        "RedrivePolicy": json.dumps({
            "deadLetterTargetArn": dlq_arn,
            "maxReceiveCount": 5
        })
    }
)

Use 3-5 as a starting point. A DLQ is not a trash bin - it's an alert.

AI-Agent-Specific Failure Modes

• Duplicate LLM calls: Bigger bill, possibly different answer. Use task_id as idempotency key.
• Non-deterministic outputs: Store first successful output.
• Tool-call side effects: Make idempotent.
• Long-running inference: Use visibility heartbeat.

What to Monitor

Two alerts:

1. Oldest message exceeds latency target
2. DLQ has messages

When SQS Is Not the Right Tool

I have seen teams burn hours building multi-agent systems with database polling and sleep timers. It works at demo scale. It usually does not survive production traffic.

SQS gives you durable message delivery primitives. But the app still needs idempotent handlers, visibility timeout tuning, and DLQ monitoring.

Default architecture:

• One queue between major stages
• Standard queues unless ordering required
• Every handler idempotent
• Large payloads outside the queue
• Visibility timeouts based on real processing time
• Dead letter queues for failures

The difference between an AI demo and a reliable AI system is rarely the prompt. It is the infrastructure around the prompt.

Build that layer intentionally.