EventBridge Pipes for AI Agents: Building a Self‑Contained Plan‑Act‑Observe Loop in Node.js

// setup-pipe.ts
import {
  EventBridgeClient,
  CreateEventBusCommand,
  CreatePipeCommand,
  TagResourceCommand,
} from "@aws-sdk/client-eventbridge";
import {
  LambdaClient,
  CreateFunctionCommand,
  AddPermissionCommand,
} from "@aws-sdk/client-lambda";
import { readFileSync } from "fs";
import { resolve } from "path";

// ---------- 1. Create a private EventBridge bus ----------
const eb = new EventBridgeClient({});
await eb.send(
  new CreateEventBusCommand({
    Name: "AgentLoopBus", // unique name inside your account
  })
);

// ---------- 2. Create the Plan‑Act‑Observe Lambda ----------
const lambda = new LambdaClient({});
await lambda.send(
  new CreateFunctionCommand({
    FunctionName: "PlanActObserve",
    Runtime: "nodejs22.x", // latest runtime as of 2026
    Role: "arn:aws:iam::123456789012:role/AgentLambdaRole", // pre‑created IAM role
    Handler: "index.handler",
    Code: {
      // zip file that contains index.js (we’ll write it later)
      ZipFile: readFileSync(resolve(__dirname, "lambda.zip")),
    },
    // SnapStart is disabled because we need VPC access for Claude (edge case)
    SnapStart: { ApplyOn: "None" },
  })
);

// ---------- 3. Allow EventBridge to invoke the Lambda ----------
await lambda.send(
  new AddPermissionCommand({
    FunctionName: "PlanActObserve",
    StatementId: "AllowEventBridgeInvoke",
    Action: "lambda:InvokeFunction",
    Principal: "events.amazonaws.com",
    // SourceArn limits the permission to our specific pipe (we’ll fill later)
    SourceArn: "arn:aws:events:us-east-1:123456789012:pipe/AgentPipe",
  })
);

// ---------- 4. Define the Pipe ----------
await eb.send(
  new CreatePipeCommand({
    Name: "AgentPipe",
    RoleArn: "arn:aws:iam::123456789012:role/AgentPipeRole", // IAM role that the pipe assumes
    Source: {
      // The same bus we created; the pipe will listen for events with detail-type = "AgentStep"
      EventBridge: {
        // 5‑second filter limit – keep it simple!
        FilterCriteria: {
          Filters: [
            {
              Pattern: JSON.stringify({
                "detail-type": ["AgentStep"],
              }),
            },
          ],
        },
        // SourceArn points to the bus we made
        Arn: "arn:aws:events:us-east-1:123456789012:event-bus/AgentLoopBus",
      },
    },
    Target: {
      // Target is the Lambda we just created
      LambdaFunction: {
        Arn: "arn:aws:lambda:us-east-1:123456789012:function:PlanActObserve",
      },
    },
    // Optional: dead‑letter queue (DLQ) for failed deliveries
    DeadLetterConfig: {
      Arn: "arn:aws:sqs:us-east-1:123456789012:AgentPipeDLQ",
    },
    // Retry policy: 3 attempts, exponential back‑off
    RetryPolicy: {
      MaximumRetryAttempts: 3,
      MaximumEventAgeInSeconds: 60,
    },
  })
);

console.log("Pipe and resources created – the loop is ready to run!");

{
  "id": "abcd‑1234",
  "detail-type": "AgentStep",
  "detail": {
    "step": 3,
    "state": { "counter": 7 },
    "observation": "API returned 200"
  }
}

// index.js – Lambda handler
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";

/**
 * The Lambda entry point.
 * @param {object} event – EventBridge event that triggered the Lambda.
 * @returns {object} – Simple status payload.
 */
export const handler = async (event) => {
  // 1️⃣ Extract useful bits from the incoming event
  const { step, state, observation } = event.detail;
  console.log("Received step:", step, "state:", state, "observation:", observation);

  // 2️⃣ Build a prompt for Claude. We ask Claude to return JSON with `action` and `payload`.
  const prompt = `
You are an autonomous planning agent. Given the current state and the last observation,
produce the next action in JSON format with two fields:
  "action": one of ["call-api", "wait", "finish"]
  "payload": an object that contains the data needed for the action.

State: ${JSON.stringify(state)}
Observation: ${observation}
Step: ${step}
`;

  // 3️⃣ Call Claude's /v1/complete endpoint.
  //    The endpoint expects a POST with a JSON body.
  const response = await fetch("https://api.anthropic.com/v1/complete", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "x-api-key": process.env.CLAUDE_API_KEY, // stored in Lambda env vars
      "anthropic-version": "2023-06-01",
    },
    body: JSON.stringify({
      model: "claude-3-5-sonnet-20240610",
      prompt: prompt,
      max_tokens_to_sample: 200,
    }),
  });

  // 4️⃣ Convert the response to JSON and guard against malformed output
  const raw = await response.text(); // keep the raw text for debugging
  let parsed;
  try {
    parsed = JSON.parse(raw);
  } catch (e) {
    console.error("Failed to parse Claude response:", raw);
    // Publish a failure event so the loop can decide what to do next
    await publishEvent(step, state, "parse_error", raw);
    throw e; // let the pipe’s retry policy handle the error
  }

  // Expected shape: { completion: "...JSON string..." }
  let actionObj;
  try {
    actionObj = JSON.parse(parsed.completion);
  } catch (e) {
    console.error("Claude did not return valid JSON:", parsed.completion);
    await publishEvent(step, state, "invalid_json", parsed.completion);
    throw e;
  }

  console.log("Claude suggested action:", actionObj);

  // 5️⃣ Prepare the next event payload
  const nextDetail = {
    step: step + 1,
    state: { ...state, lastAction: actionObj.action }, // simple state update
    observation: `Action ${actionObj.action} queued`,
  };

  // 6️⃣ Send the new event back to the same bus, same detail-type.
  await publishEvent(nextDetail.step, nextDetail.state, "action_queued", nextDetail);

  // Lambda must return something; the pipe ignores it.
  return { status: "ok" };
};

/**
 * Helper that writes an event to the EventBridge bus used by the pipe.
 * @param {number} step
 * @param {object} state
 * @param {string} observation
 * @param {object} detail
 */
async function publishEvent(step, state, observation, detail) {
  const eb = new EventBridgeClient({});
  const command = new PutEventsCommand({
    Entries: [
      {
        EventBusName: "AgentLoopBus",
        Source: "my.agent",
        DetailType: "AgentStep",
        Time: new Date(),
        Detail: JSON.stringify({
          step,
          state,
          observation,
          // Preserve any extra fields the caller gave us
          ...(detail || {}),
        }),
      },
    ],
  });

  const result = await eb.send(command);
  console.log("Published next step:", result);
}

// Inside the fetch block – replace the previous fetch call with this
const response = await fetch("https://api.anthropic.com/v1/complete", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "x-api-key": process.env.CLAUDE_API_KEY,
    "anthropic-version": "2023-06-01",
  },
  body: JSON.stringify({
    model: "claude-3-5-sonnet-20240610",
    prompt: prompt,
    max_tokens_to_sample: 200,
  }),
});

if (!response.ok) {
  const errorBody = await response.text();
  console.error(`Claude API error ${response.status}:`, errorBody);
  // Publish a special “api_error” event so the loop can decide to back‑off
  await publishEvent(step, state, "api_error", { status: response.status, body: errorBody });
  // Throw to trigger the pipe’s retry policy
  throw new Error(`Claude API responded with ${response.status}`);
}

import { SQSClient, CreateQueueCommand, SetQueueAttributesCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});
// 1️⃣ Create the queue
const { QueueUrl } = await sqs.send(
  new CreateQueueCommand({
    QueueName: "AgentPipeDLQ",
    Attributes: {
      // Enable content‑based deduplication if you want exactly‑once semantics
      MessageDeduplicationId: "true",
    },
  })
);

// 2️⃣ Allow EventBridge to write to the queue
await sqs.send(
  new SetQueueAttributesCommand({
    QueueUrl,
    Attributes: {
      Policy: JSON.stringify({
        Version: "2012-10-17",
        Statement: [
          {
            Effect: "Allow",
            Principal: { Service: "events.amazonaws.com" },
            Action: "sqs:SendMessage",
            Resource: `arn:aws:sqs:${process.env.AWS_REGION}:${process.env.AWS_ACCOUNT_ID}:AgentPipeDLQ`,
          },
        ],
      }),
    },
  })
);

// Example: more aggressive retry
RetryPolicy: {
  MaximumRetryAttempts: 5,
  MaximumEventAgeInSeconds: 120, // give the loop up to 2 minutes to finish a step
},