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

// 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;

// 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);

// 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);
};

// 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);
};

// 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);
};

// 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);

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

// 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);

// 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);