EventBridge Pipes Filter Expressions

Filter pattern structure

A Pipes filter pattern is structurally identical to an EventBridge event pattern on a bus rule. The pattern is a JSON object where each key corresponds to a path in the source record, and the value is an array of match conditions (array = OR for that field's conditions).

For SQS sources, the record structure that the filter sees is:

{
  "messageId": "abc-123",
  "receiptHandle": "...",
  "body": "{\"eventType\":\"tool_call\",\"tenantId\":\"t-99\",\"toolName\":\"search\"}",
  "attributes": {
    "ApproximateReceiveCount": "1",
    "SentTimestamp": "1727126400000"
  },
  "messageAttributes": {
    "tenantId": { "stringValue": "t-99", "dataType": "String" },
    "priority": { "stringValue": "high", "dataType": "String" }
  },
  "eventSource": "aws:sqs",
  "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:mcp-tool-calls"
}

Note that body is a string — the raw SQS message body. EventBridge Pipes automatically parses it as JSON when evaluating filter patterns against $.body.* fields. You do not need to pre-parse it; the filter engine treats it as a nested object for pattern matching purposes.

AND vs OR logic

This is the most common source of filtering bugs:

• Multiple conditions on different fields within a single filter pattern → AND. All conditions must match for the record to be passed. Example: {"body": {"eventType": ["tool_call"], "priority": ["high"]}} passes only records where both eventType is tool_call AND priority is high.
• Multiple conditions on the same field in the array → OR for that field. Example: {"body": {"eventType": ["tool_call", "tool_result"]}} passes records where eventType is tool_call OR tool_result.
• Multiple patterns in the filters list → OR across patterns. A record is passed if it matches any pattern in the list.

// CDK — filter patterns demonstrating AND vs OR logic
const pipe = new pipes.CfnPipe(this, "RoutingPipe", {
  // ...
  sourceParameters: {
    sqsQueueParameters: { batchSize: 10 },
    filterCriteria: {
      filters: [
        // Pattern 1: eventType = tool_call AND priority = high (AND logic)
        {
          pattern: JSON.stringify({
            body: {
              eventType: ["tool_call"],
              priority: ["high"],
            },
          }),
        },
        // Pattern 2: eventType = tool_result (separate pattern = OR with Pattern 1)
        {
          pattern: JSON.stringify({
            body: {
              eventType: ["tool_result"],
            },
          }),
        },
      ],
      // Combined effect: pass if (eventType=tool_call AND priority=high) OR (eventType=tool_result)
    },
  },
  // ...
});

Full operator reference

Nested JSON with dot-notation

For deeply nested fields in the SQS body, use the natural JSON nesting structure in the filter pattern — not dot-notation string keys. EventBridge Pipes (like EventBridge Rules) uses nested object matching, not JSONPath dot-notation in the key name.

// Tool call event body structure (nested)
// {
//   "event": {
//     "type": "tool_call",
//     "context": {
//       "tenant": { "id": "t-99", "plan": "team" }
//     }
//   }
// }

// CORRECT — nested object matching
const correctFilter = {
  body: {
    event: {
      type: ["tool_call"],
      context: {
        tenant: {
          plan: ["team", "enterprise"],  // OR: team or enterprise plan
        },
      },
    },
  },
};

// WRONG — dot-notation string keys don't work in filter patterns
const wrongFilter = {
  "body.event.type": ["tool_call"],  // This never matches anything
};

// CDK usage
sourceParameters: {
  filterCriteria: {
    filters: [{ pattern: JSON.stringify(correctFilter) }],
  },
}

Filtering on SQS message attributes

SQS message attributes are separate from the message body and are accessed via a different path in the filter pattern. This is commonly used for multi-tenant MCP server architectures where the tenant ID is set as a message attribute by the producer, allowing routing without parsing the body.

// SQS producer: set tenantId as a message attribute
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({ region: "us-east-1" });
await sqs.send(new SendMessageCommand({
  QueueUrl: process.env.QUEUE_URL,
  MessageBody: JSON.stringify({ toolName: "search", query: "..." }),
  MessageAttributes: {
    tenantId: {
      DataType: "String",
      StringValue: "t-99",
    },
    priority: {
      DataType: "String",
      StringValue: "high",
    },
  },
}));

// Pipes filter: match on message attribute
const attrFilter = {
  messageAttributes: {
    // Filter on tenantId attribute
    tenantId: {
      stringValue: ["t-99", "t-100"],  // OR: either tenant
    },
    // AND filter on priority
    priority: {
      stringValue: ["high"],
    },
  },
};

// Combined body + attribute filter
const combinedFilter = {
  body: {
    eventType: ["tool_call"],
  },
  messageAttributes: {
    priority: { stringValue: ["high"] },
  },
};

Important: the message attribute filter syntax uses the attribute name as a nested key with stringValue, numberValue, or binaryValue properties matching the DataType set by the producer. Binary attributes (binaryValue) are base64-encoded strings in filter patterns.

Filter vs enrichment Lambda — when filtering isn't enough

Filter expressions evaluate static pattern matching. They cannot:

• Call external services (DynamoDB, SSM Parameter Store) to look up whether a tenant is allowed
• Apply complex business logic (e.g., rate-limit checks based on event history)
• Parse non-JSON body formats (XML, CSV, custom binary)
• Access or compare fields from different events in the batch

For these cases, use the enrichment Lambda as a filter: return an empty array [] from the enrichment function for records that should be dropped. EventBridge Pipes treats an empty enrichment response as "no records to pass to target" for that batch — effectively filtering them out.

// Enrichment Lambda as dynamic filter + enricher
exports.handler = async (events) => {
  const tenantCache = new Map();

  const enriched = await Promise.all(events.map(async (evt) => {
    const body = JSON.parse(evt.body);

    // Dynamic check: look up tenant plan in DynamoDB
    const tenantId = body.tenantId;
    if (!tenantCache.has(tenantId)) {
      const result = await dynamodb.getItem({
        TableName: "tenants",
        Key: { tenantId: { S: tenantId } },
      }).promise();
      tenantCache.set(tenantId, result.Item?.plan?.S ?? "free");
    }

    const plan = tenantCache.get(tenantId);

    // Drop free-tier events from this high-priority pipeline
    if (plan === "free") return null;

    return { ...body, tenantPlan: plan, enrichedAt: new Date().toISOString() };
  }));

  // Filter out dropped records (null values)
  // Pipes treats empty array response as "pass 0 records to target"
  return enriched.filter(Boolean);
};

Common filter pattern mistakes