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.
Why EventBridge Pipes Matter for AI Agents
When you build an autonomous agent you need three moving parts:
- Plan – ask a large language model (LLM) what to do next.
- Act – run a tool or service based on the plan.
- Observe – read the result and feed it back to the planner.
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:
- automatic retries up to the limit you set,
- built‑in observability via CloudWatch metrics,
- horizontal scaling without writing extra code,
- a clear visual diagram in the console that shows the data flow.
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.
Quick vocabulary
- EventBridge – a serverless event bus that lets different AWS services talk to each other.
- Pipe – a declarative link that says “when an event matching X arrives, send it to Y.”
- Source – the service that creates the event (here, the same Pipe feeding back to itself).
- Target – the service that receives the event (our Plan‑Act‑Observe Lambda).
Setting Up the Pipe: Resources and Permissions
Before we write any code we need the infrastructure that lets the pipe move events safely. The steps are:
- Create an EventBridge bus – a private channel for our agent’s events.
- Create a Lambda function that will serve as the single place to plan, act, and observe.
- Grant the Pipe permission to invoke the Lambda and to publish events back to the bus.
- Define the Pipe with a source (the bus) and a target (the Lambda).
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 the
TestEventPatternconsole tool first.
Gotchas to keep in mind
| 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.
Implementing the Plan‑Act‑Observe Lambda
The Lambda is the only piece of custom code we write. Its responsibilities are:
- Read the incoming event – it contains the current state and the last observation.
- Call Claude (the LLM) with a prompt that includes the state and observation.
- Parse Claude’s JSON response to extract the next action (e.g., “call‑api”, “wait”, “finish”).
- Publish a new event back to the same EventBridge bus so the pipe can feed it into the next iteration.
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 does one thing – translate a step into a new event. All retry, scaling, and delivery concerns are handled by the pipe.
Why keep the Lambda tiny?
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.
Connecting Claude via HTTP and Handling Responses
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.
// 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
LambdaInvocationErrorsmetric. Pair that with a CloudWatch alarm on the pipe’sDeliveryFailedmetric to get early alerts.
Analogy for the response handling
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.
Observability, Retries, and Dead‑Letter Queues
Even with a pipe that retries automatically, you still need visibility into why a particular iteration stopped. AWS gives you three built‑in tools:
-
CloudWatch Metrics –
DeliveryAttempts,DeliveryFailed,AgeOfOldestMessage. - EventBridge DLQ (Dead‑Letter Queue) – an SQS queue that receives events that could not be delivered after all retries.
- Lambda Destinations – a way to route successful or failed invocations to other services (optional for extra logging).
Wiring a DLQ
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.
Configuring the retry policy
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
},
Observability checklist
| 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.
The Takeaway
You now have a complete, production‑ready loop that runs entirely on EventBridge Pipes and a single Lambda.
- EventBridge Pipes replace custom “glue” Lambda code with a managed, observable connection that automatically retries and scales.
- A tiny Plan‑Act‑Observe Lambda does three things: read the incoming event, call Claude, and publish the next event.
- HTTP calls to Claude need correct headers and defensive JSON parsing; otherwise the pipe will keep retrying without progress.
- Configure a dead‑letter queue and a clear retry policy; otherwise a single error will halt the whole loop.
- Keep filter patterns under the 5‑second limit and store large data outside the event payload to avoid silent drops.
- Use CloudWatch metrics, alarms, and DLQ monitoring to stay aware of failures and latency spikes.
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 notice
This article was written with the help of an AI system — Groq (GPT OSS 120B).
Published: 2026-08-25 · Primary focus: EventBridge
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)