DEV Community

Dinesh_gowtham
Dinesh_gowtham

Posted on

AI Agents with SQS and Lambda: Build a Plan‑Act‑Observe Loop in Node.js

#aiagents #sqs #lambda #typescript

You’ve seen flashy EventBridge Pipes demos, but they hide the mechanics of an autonomous AI agent. In just a few lines of TypeScript you can wire SQS and Lambda together to give your LLM a reliable tool‑use loop. Let’s demystify the messaging backbone that makes the agent think, act, and observe.

The Plan‑Act‑Observe Loop Made Simple

Why a loop?

Think of an autonomous agent like a chef following a recipe:

  1. Plan – decide what ingredient (tool) is needed.
  2. Act – go to the pantry (external API) and fetch it.
  3. Observe – taste the dish (read the result) and decide the next step.

Repeating these three steps lets the LLM keep a conversation alive, call APIs, and adjust its next prompt based on real data.

Key terms

  • LLM – a large language model that generates text, e.g., OpenAI’s GPT‑4.
  • Tool request – a structured JSON payload the LLM emits when it wants to call an external service.
  • Message – a small packet of data that travels through Amazon SQS (Simple Queue Service).

Below is a tiny TypeScript type that captures a tool request. Using the satisfies keyword tells the compiler “this object must match the shape, but keep the exact literal types for later safety.”

// src/types.ts
export interface ToolRequest {
  /** Unique identifier for the step – used for deduplication */
  requestId: string;
  /** Name of the tool the LLM wants to use, e.g. "weather" */
  toolName: string;
  /** Arbitrary parameters the tool needs, kept as a plain object */
  args: Record<string, unknown>;
}

/* The `satisfies` operator checks that the literal we export conforms to ToolRequest
   without widening the type – helpful for strict runtime validation later. */
export const exampleRequest = {
  requestId: "req-001",
  toolName: "weather",
  args: { location: "Seattle, WA" },
} satisfies ToolRequest;
Enter fullscreen mode Exit fullscreen mode

In plain English – The loop is just the chef’s three‑step routine. By breaking the job into “plan, act, observe” we give the LLM a predictable place to read, write, and react to data.

Creating a FIFO SQS Queue for Deterministic Agent Steps

Why FIFO?

A FIFO (First‑In‑First‑Out) queue guarantees that messages are processed in the exact order they were sent. For a planner‑executor‑observer chain, out‑of‑order execution can corrupt the reasoning flow (imagine adding salt before the broth is even simmered).

How to create it – we’ll use the @aws-sdk/client-sqs v3 client. The code below can be run locally with the AWS CLI configured, or as part of a CDK deployment script.

// src/createQueue.ts
import {
  SQSClient,
  CreateQueueCommand,
  CreateQueueCommandInput,
} from "@aws-sdk/client-sqs";

// The SQS client reads credentials from the environment (AWS_ACCESS_KEY_ID, etc.)
const sqs = new SQSClient({});

/** Create a FIFO queue named "agent-steps.fifo". */
async function createFifoQueue() {
  const params: CreateQueueCommandInput = {
    QueueName: "agent-steps.fifo", // FIFO queues must end with .fifo
    Attributes: {
      // Guarantees exactly‑once processing when combined with content‑based deduplication
      FifoQueue: "true",
      // Enable content‑based deduplication so we can send the same payload twice
      // without creating duplicate entries (useful for retries)
      ContentBasedDeduplication: "true",
      // Visibility timeout: how long a message stays invisible after a Lambda receives it
      VisibilityTimeout: "30", // seconds – must be longer than Lambda execution time
      // Long polling reduces empty receives (costs) by waiting up to 20 seconds
      ReceiveMessageWaitTimeSeconds: "20",
    },
  };

  const command = new CreateQueueCommand(params);
  const response = await sqs.send(command);
  console.log("FIFO queue URL:", response.QueueUrl);
}

createFifoQueue().catch(console.error);
Enter fullscreen mode Exit fullscreen mode

Tip – The VisibilityTimeout must be longer than the longest Lambda execution that reads from this queue. If it’s shorter, the same message can become visible again while the first Lambda is still working, leading to duplicate external calls.

Lambda Handlers: Planner, Executor, and Observer

Why three Lambdas?

Separating responsibilities keeps each function small, testable, and easier to reason about.

  • Planner reads a tool request from the plan queue, decides which executor to invoke, and forwards the request.
  • Executor actually calls the external API (e.g., a weather service) and pushes the raw response onto the response queue.
  • Observer pulls the response, formats it for the LLM, and places the next prompt back onto the plan queue.

Planner Lambda 

// src/plannerLambda.ts
import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  SendMessageCommand,
} from "@aws-sdk/client-sqs";
import {
  LambdaClient,
  InvokeCommand,
} from "@aws-sdk/client-lambda";
import { ToolRequest } from "./types";

const sqs = new SQSClient({});
const lambda = new LambdaClient({});

const PLAN_QUEUE_URL = process.env.PLAN_QUEUE_URL!;
const EXECUTOR_FUNCTION = process.env.EXECUTOR_FUNCTION!;

/** Entry point for the Planner Lambda */
export const handler = async (): Promise<void> => {
  // Pull one message at a time to keep ordering intact
  const receive = new ReceiveMessageCommand({
    QueueUrl: PLAN_QUEUE_URL,
    MaxNumberOfMessages: 1,
    WaitTimeSeconds: 20, // long polling
    VisibilityTimeout: 30, // seconds – matches queue attribute
  });

  const { Messages } = await sqs.send(receive);
  if (!Messages?.length) return; // nothing to do

  const raw = Messages[0];
  const body = JSON.parse(raw.Body!) as ToolRequest;

  // Forward the request to the Executor Lambda
  const invoke = new InvokeCommand({
    FunctionName: EXECUTOR_FUNCTION,
    Payload: Buffer.from(JSON.stringify(body)),
    // Invoke synchronously so we can delete the message only after success
    InvocationType: "RequestResponse",
  });
  await lambda.send(invoke);

  // Remove the message now that processing succeeded
  const del = new DeleteMessageCommand({
    QueueUrl: PLAN_QUEUE_URL,
    ReceiptHandle: raw.ReceiptHandle!,
  });
  await sqs.send(del);
};
Enter fullscreen mode Exit fullscreen mode

Executor Lambda 

// src/executorLambda.ts
import {
  SQSClient,
  SendMessageCommand,
} from "@aws-sdk/client-sqs";
import fetch from "node-fetch"; // native fetch works in Node 22, but keep explicit for clarity
import { ToolRequest } from "./types";

const sqs = new SQSClient({});
const RESPONSE_QUEUE_URL = process.env.RESPONSE_QUEUE_URL!;

/** Simple executor that knows only how to call a weather API */
export const handler = async (event: any): Promise<void> => {
  // The event payload is the ToolRequest JSON string from Planner
  const request: ToolRequest = JSON.parse(event.body?.toString() ?? event);

  // Very small example – a real implementation would handle errors, auth, etc.
  const apiUrl = `https://api.open-meteo.com/v1/forecast?latitude=47.61&longitude=-122.33&current_weather=true`;
  const apiResponse = await fetch(apiUrl);
  const data = await apiResponse.json();

  // Package the raw API response together with the original requestId
  const responseMessage = {
    requestId: request.requestId,
    toolName: request.toolName,
    result: data,
  };

  // Push the result onto the response queue for the Observer
  const send = new SendMessageCommand({
    QueueUrl: RESPONSE_QUEUE_URL,
    MessageBody: JSON.stringify(responseMessage),
    MessageGroupId: "responses", // required for FIFO queues
    MessageDeduplicationId: request.requestId, // deduplicate retries
  });
  await sqs.send(send);
};
Enter fullscreen mode Exit fullscreen mode

Observer Lambda 

// src/observerLambda.ts
import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  SendMessageCommand,
} from "@aws-sdk/client-sqs";
import { ToolRequest } from "./types";

const sqs = new SQSClient({});
const RESPONSE_QUEUE_URL = process.env.RESPONSE_QUEUE_URL!;
const PLAN_QUEUE_URL = process.env.PLAN_QUEUE_URL!;

/** Reads the API result, formats a new LLM prompt, and puts it back on the plan queue */
export const handler = async (): Promise<void> => {
  const receive = new ReceiveMessageCommand({
    QueueUrl: RESPONSE_QUEUE_URL,
    MaxNumberOfMessages: 1,
    WaitTimeSeconds: 20,
    VisibilityTimeout: 30,
  });

  const { Messages } = await sqs.send(receive);
  if (!Messages?.length) return;

  const raw = Messages[0];
  const payload = JSON.parse(raw.Body!);

  // Create a friendly LLM prompt that includes the observed data
  const nextPrompt = {
    requestId: payload.requestId,
    toolName: "continue", // special token that tells the LLM to keep going
    args: {
      observation: payload.result,
      instruction: "Summarize the weather and decide if we need an umbrella.",
    },
  } satisfies ToolRequest;

  // Send the new request back to the planner queue
  const send = new SendMessageCommand({
    QueueUrl: PLAN_QUEUE_URL,
    MessageBody: JSON.stringify(nextPrompt),
    MessageGroupId: "plans",
    MessageDeduplicationId: nextPrompt.requestId,
  });
  await sqs.send(send);

  // Delete the processed response message
  const del = new DeleteMessageCommand({
    QueueUrl: RESPONSE_QUEUE_URL,
    ReceiptHandle: raw.ReceiptHandle!,
  });
  await sqs.send(del);
};
Enter fullscreen mode Exit fullscreen mode

Helpful tip – Keep each Lambda under 10 seconds for this demo. If you need longer processing, increase the queue’s VisibilityTimeout accordingly, and remember the gotcha about duplicates (see the next section).

Connecting the Pieces with the AWS SDK v3 and TypeScript ‘satisfies’

Why the v3 SDK?

Version 3 of the AWS SDK ships each service as a separate, tree‑shakable package (@aws-sdk/client-sqs, @aws-sdk/client-lambda). This reduces bundle size for Lambda layers and makes the import graph clearer for newcomers.

How to wire everything together – a tiny “bootstrap” script that seeds the first plan message and shows the flow end‑to‑end.

// src/seed.ts
import {
  SQSClient,
  SendMessageCommand,
} from "@aws-sdk/client-sqs";
import { ToolRequest } from "./types";

const sqs = new SQSClient({});
const PLAN_QUEUE_URL = process.env.PLAN_QUEUE_URL!;

/** Kick‑starts the loop with an initial tool request */
async function seedPlan() {
  const initialRequest: ToolRequest = {
    requestId: "req-" + Date.now(),
    toolName: "weather",
    args: { location: "Seattle, WA" },
  };

  const cmd = new SendMessageCommand({
    QueueUrl: PLAN_QUEUE_URL,
    MessageBody: JSON.stringify(initialRequest),
    MessageGroupId: "plans",
    // Using the requestId as deduplication id prevents accidental re‑queues
    MessageDeduplicationId: initialRequest.requestId,
  });

  const res = await sqs.send(cmd);
  console.log("Seeded plan message, MessageId:", res.MessageId);
}

seedPlan().catch(console.error);
Enter fullscreen mode Exit fullscreen mode

Key TypeScript pattern – satisfies

const nextPrompt = {
  requestId: payload.requestId,
  toolName: "continue",
  args: { /* ... */ },
} satisfies ToolRequest;
Enter fullscreen mode Exit fullscreen mode

The satisfies keyword makes sure nextPrompt adheres to the ToolRequest shape without widening the literal types. This gives us compile‑time safety (the LLM never receives a malformed payload) while preserving exact string literals for downstream JSON schema checks.

In plain English – Think of the SDK as a set of toolboxes (SQS, Lambda). You pick the exact toolbox you need, and TypeScript’s satisfies is like a checklist that guarantees every tool you put in the box matches the required specification.

Observability, Retries, and the Visibility‑Timeout Gotcha

Why observability matters

When an autonomous agent runs unattended, you need to know whether each step succeeded, failed, or was retried. CloudWatch metrics, log statements, and DLQ (Dead‑Letter Queue) wiring give you that visibility.

The visibility‑timeout pitfall – If a Lambda takes 45 seconds but the queue’s visibility timeout is 30 seconds, the message reappears after 30 seconds. The Lambda may still be finishing its work, so the same request gets handed to a second Lambda instance. The external API is called twice, which looks like the agent “hallucinated” an extra action.

Fixing the problem 

// src/updateTimeout.ts
import {
  SQSClient,
  SetQueueAttributesCommand,
} from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});
const PLAN_QUEUE_URL = process.env.PLAN_QUEUE_URL!;

async function extendVisibility() {
  const cmd = new SetQueueAttributesCommand({
    QueueUrl: PLAN_QUEUE_URL,
    Attributes: {
      // Make the timeout 90 seconds – comfortably larger than any Lambda in this demo
      VisibilityTimeout: "90",
    },
  });
  await sqs.send(cmd);
  console.log("Visibility timeout updated to 90 seconds");
}

extendVisibility().catch(console.error);
Enter fullscreen mode Exit fullscreen mode

Adding a DLQ for permanent failures 

// src/createDlq.ts
import {
  SQSClient,
  CreateQueueCommand,
} from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});

async function createDlq() {
  // Simple standard queue to collect messages that exceeded max retries
  const dlq = await sqs.send(
    new CreateQueueCommand({ QueueName: "agent-dlq" })
  );

  const planQueue = await sqs.send(
    new CreateQueueCommand({
      QueueName: "agent-steps.fifo",
      Attributes: {
        FifoQueue: "true",
        RedrivePolicy: JSON.stringify({
          deadLetterTargetArn: dlq.QueueArn,
          maxReceiveCount: "5", // after 5 attempts, move to DLQ
        }),
      },
    })
  );

  console.log("DLQ URL:", dlq.QueueUrl);
  console.log("Plan queue URL:", planQueue.QueueUrl);
}
createDlq().catch(console.error);
Enter fullscreen mode Exit fullscreen mode

Logging pattern – Insert a single console.log at the start and end of each Lambda, including requestId. CloudWatch will automatically group logs by request ID, making it easy to trace a full plan‑act‑observe cycle.

Key takeaway – Always set the SQS visibility timeout longer than the Lambda’s maximum execution time. Pair that with a DLQ and you’ll avoid duplicate external calls and have a clean audit trail.

The Takeaway

  • A FIFO SQS queue guarantees ordered delivery, which is essential for the deterministic “plan‑act‑observe” workflow.
  • Splitting responsibilities into three tiny Lambdas (Planner, Executor, Observer) keeps each function simple and easier to test.
  • The AWS SDK v3’s modular packages (@aws-sdk/client-sqs, @aws-sdk/client-lambda) reduce bundle size and make imports clearer for beginners.
  • Using TypeScript’s satisfies operator validates message shapes at compile time without losing literal types.
  • The visibility‑timeout must exceed the longest Lambda execution; otherwise duplicate processing creates the illusion of “hallucinated” actions.
  • Attach a Dead‑Letter Queue and consistent log statements to surface failures early and keep the autonomous loop trustworthy.

With these building blocks you can assemble a reliable AI‑agent loop that’s transparent, debuggable, and cost‑predictable—no EventBridge Pipes required. Happy coding!

Transparency notice

This article was written with the help of an AI system — Groq (GPT OSS 120B).

Published: 2026-08-24 · Primary focus: SQS

All code blocks are intended to be correct and runnable, but please verify them
against the official docs for the tools mentioned before using in production.

Find an error? Drop a comment — corrections are always welcome.

Top comments (0)