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

A developer demonstrates how to build a self-contained plan-act-observe loop for AI agents using AWS EventBridge Pipes in Node.js, eliminating the need for custom glue Lambda code. The approach uses declarative wiring to connect LLM calls, tool invocations, and feedback loops, with managed retries and observability.

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. 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 " ; Tip:Keep the filter pattern under 5 seconds of evaluation time. Complex JSONPath expressions will be dropped silently, so test them with the TestEventPattern console tool first. | Service | Gotcha | How to avoid | |---|---|---| | EventBridge Pipes | 5‑second filter evaluation limit | Use simple key/value matching; avoid deep nesting. | | EventBridge Scheduler | Timezone handling around DST | Always store timestamps in UTC and convert only for display. | | Schema Registry | Events must flow once before the schema is inferred | Publish a “warm‑up” event before the first real iteration. | | Cross‑account routing | Resource‑based policies are easy to misconfigure | Grant events:PutEvents on the target account explicitly. | | Delivery delay under high load | Can reach 30+ seconds | Design your agent to be tolerant of a few seconds of latency. | 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. js // 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: | Pitfall | What happens | Fix | |---|---|---| Missing Content-Type: application/json header | Claude returns a generic HTML error page, which later fails JSON parsing. | Always set Content-Type to application/json . | Not setting anthropic-version header | API returns a 400 with a message about outdated version. | Include the header with the current version e.g., 2023-06-01 . | | Large JSON payloads exceeding Claude’s 5 MB limit | The request is silently dropped, and the pipe’s retry policy eventually gives up. | Keep the prompt under a few kilobytes; store big data elsewhere S3 and pass a reference. | 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. js // 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 a LambdaInvocationErrors metric. Pair that with a CloudWatch alarm on the pipe’s DeliveryFailed 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 pause. 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. js 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 }, | Item | Why it matters | How to enable | |---|---|---| CloudWatch alarm on DeliveryFailed 0 | Detects a stuck agent early | Create an alarm that notifies Slack or email | Lambda Duration metric | Shows if a step takes longer than expected maybe a slow external API | Add a CloudWatch dashboard widget | | DLQ monitoring | Captures events that fell through all retries | Set up a Lambda that writes DLQ messages to a log file or alerts | EventBridge AgeOfOldestMessage | If the pipe is backing up, the age will rise | Add a threshold alarm e.g., 30 seconds | 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.