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

Imagine an AI agent that can plan, act, and observe without a single Lambda function acting as glue. EventBridge Pipes let you stitch together LLM calls, tool invocations, and feedback loops with declarative wiring. This post shows exactly how to wire that up.

When you build an autonomous agent you need three moving parts:

The naïve approach is to write one Lambda that does all three steps or to chain several Lambdas with manual invoke

calls. That creates a lot of “glue code”: retry logic, error handling, scaling decisions, and metrics are all your responsibility.

EventBridge Pipes are a managed way to connect a source (an EventBridge event) directly to a target (a Lambda, an HTTP endpoint, another EventBridge bus, etc.). The pipe evaluates a simple filter, optionally transforms the payload, and then delivers it. Because the service is built into EventBridge you get:

Think of a pipe as a conveyor belt in a factory. The belt moves a product (your event) from one station (the planner) to the next (the actor) without a worker having to pick it up, carry it, and drop it again. If the belt breaks, the factory’s alarm system (CloudWatch) tells you instantly, and the belt can try to move the product again automatically.

In plain English:EventBridge Pipes replace custom “glue” Lambda code with a managed, observable connection that retries for you.

Before we write any code we need the infrastructure that lets the pipe move events safely. The steps are:

All of this can be done with the AWS SDK for JavaScript (v3). Below is a minimal script you can run locally or in a CI job. It uses the @aws-sdk/client-eventbridge

and @aws-sdk/client-lambda

packages you requested.

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

Tip:Keep the filter pattern under 5 seconds of evaluation time. Complex JSONPath expressions will be dropped silently, so test them with theTestEventPattern

console tool first.

In plain English:The pipe itself does most of the heavy lifting, but you still need a tiny amount of IAM plumbing and a very simple filter.

The Lambda is the only piece of custom code we write. Its responsibilities are:

Because the Lambda is invoked by the pipe, it receives an event

object that looks like:

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

Below is a fully commented implementation. It uses only the standard fetch

API (available in Node 22) and the @aws-sdk/client-eventbridge

client to publish the next event.

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

Key takeaway:The Lambda doesonething – translate a step into a new event. All retry, scaling, and delivery concerns are handled by the pipe.

A small function is quicker to cold‑start, cheaper to run, and easier to reason about. When the agent loop is the only thing the Lambda does, you avoid the “Lambda glue” anti‑pattern that most teams fall into.

Calling an LLM over HTTP looks straightforward, but there are three hidden pitfalls that trip up beginners:

The Lambda code above already includes the correct headers. The next piece is robust error handling. If Claude returns a non‑2xx status we want to surface that as an observable metric rather than let the pipe think the Lambda succeeded.

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

Tip:CloudWatch automatically creates aLambdaInvocationErrors

metric. Pair that with a CloudWatch alarm on the pipe’sDeliveryFailed

metric to get early alerts.

Think of Claude as a remote kitchen. You send a recipe (the prompt) and expect a plated dish (JSON). If the kitchen sends back a “Sorry, we’re closed” (HTTP 4xx) or a burnt dish (malformed JSON), you need to decide whether to try again later or change the recipe. The Lambda’s error‑handling code is your “waiter” that reports the problem back to the manager (the pipe) so the system can retry or .

Even with a pipe that retries automatically, you still need visibility into why a particular iteration stopped. AWS gives you three built‑in tools:

DeliveryAttempts

, DeliveryFailed

, AgeOfOldestMessage

. When we created the pipe we added a DeadLetterConfig

. The queue must exist before the pipe is created, and it needs a policy that allows EventBridge to send messages.

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`,
          },
        ],
      }),
    },
  })
);

When a delivery finally fails (for example, because the Lambda consistently throws a parsing error), the original event lands in the DLQ. You can set up a Lambda consumer on that queue to alert the team, store the bad event for later analysis, or even re‑inject it after fixing the bug.

In plain English:The DLQ is your safety net. Without it, a failed step disappears silently, and the agent loop stops.

The pipe we built uses MaximumRetryAttempts: 3

. That means EventBridge will try to invoke the Lambda up to three times with exponential back‑off (e.g., 1 s, 2 s, 4 s). If you need more resilience, increase the attempts, but remember that each retry adds to the overall latency of the loop.

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

Tip:The 5‑second filter evaluation limit means a complex filter can cause silent failures. Keep filters simple and test them with the console’s “Test pattern” tool.

You now have a complete, production‑ready loop that runs entirely on EventBridge Pipes and a single Lambda.

With these pieces in place you can build more sophisticated agents—adding tool‑specific Lambda targets, branching pipelines, or even cross‑account routing—while keeping the core loop simple, observable, and cost‑effective. Happy piping!

Transparency noticeThis article was written with the help of an AI system —

[Groq](GPT OSS 120B).

Published:2026-08-25 ·Primary focus:EventBridgeAll 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.