AI Agent Event Filters: Skip Runs Before the LLM

Put a deterministic check between the trigger and the model, and run the agent only for events that pass. In Promptise, set filter_expression on the trigger: a safe expression such as "action == 'opened' and payload.issue.user.type != 'Bot'", or a Python function that takes the event and returns a bool. Events that don't match never reach the queue, the prompt or the model. A webhook answers them with 200 ignored; other triggers count them in status()["filtered_count"].

• The expression language is a whitelist, parsed when the config is created. It never goes through eval.

• Mistakes fail early: a bad expression is a ValidationError at config time, not a crash on the first event.

• Missing keys are `None`, so filters don't break on events without the field.

• Errors at run time skip the event and log a warning, so a filter can't crash the agent.

• The savings are easy to count: the model calls you avoid equal the events you filter, times the calls per run.

An event-driven agent has a trigger (webhook, schedule, event bus, file watch, queue) that turns something outside into an event, and a worker that runs the agent once per event. A filter sits between the two. It sees the raw event and answers one question: is this event worth a run?

outside world ──▶ trigger ──▶ filter ──┬── no ──▶ skipped (webhook: 200 ignored,
                                       │                  others: filtered_count + 1)
                                       │ yes
                                       ▼
                       task_created hooks ──▶ queue ──▶ worker
                                                         │
                                       payload scan ──▶ prompt ──▶ LLM call

The order matters for cost. A skipped event costs one evaluation of a small expression: microseconds, no tokens. An event that passes costs at least one model call, and an agent with tools usually makes several per run.

In Promptise, where the filter runs depends on the trigger:

• Webhook: the trigger evaluates the filter while it handles the request, and answers 200 {"status": "ignored"} for events that don't match. The sender sees a success and won't retry.

• All other triggers (cron, event, message, file watch, custom): the process's listener evaluates it before the event is queued, and adds 1 to status()["filtered_count"].

A filter decides *relevance*, not *safety*. Payload text is still untrusted input once it reaches the agent; prompt-injection defense is a separate layer (see AI agent prompt injection defense).

Use a filter when the rule is a fact about the event. Use a model when the rule is about meaning. A common setup combines them: a filter drops the events that are obviously irrelevant, and the agent handles the rest. Don't filter on anything the sender controls if skipping matters for security; an attacker can set any field in a payload.

• Python 3.10 or later and Promptise Foundry 1.5.0 or later: pip install "promptise>=1.5.0".

• No API key: steps 1 to 3 need no model, and steps 4 and 5 use a small scripted, OpenAI-compatible server (fake_llm.py) that counts the model calls it gets.

• About 15 minutes.

Start with the rule in words: "run the triage agent for issues that are newly opened by a person and mention a crash". As an expression, test it with compile_filter, the same function the runtime uses:

"""Step 1: write a filter expression and test it against sample events (no model needed)."""
from promptise.runtime import TriggerEvent
from promptise.runtime.triggers import compile_filter

EXPRESSION = (
    "action == 'opened'"
    " and payload.issue.user.type != 'Bot'"
    " and contains(lower(payload.issue.title), 'crash')"
)
accept = compile_filter(EXPRESSION)


def issue(action: str, title: str, user_type: str = "User") -> TriggerEvent:
    return TriggerEvent(trigger_id="github", trigger_type="webhook", payload={
        "action": action,
        "issue": {"title": title, "user": {"login": "sam", "type": user_type}},
    })


samples = {
    "new crash report": issue("opened", "App CRASHES on login"),
    "edited crash report": issue("edited", "App crashes on login"),
    "bot-opened issue": issue("opened", "Crash log digest", user_type="Bot"),
    "feature request": issue("opened", "Add dark mode"),
    "ping (no issue key)": TriggerEvent(trigger_id="github", payload={"zen": "Keep it simple."}),
}
for name, event in samples.items():
    print(f"{name:<22} -> {'run the agent' if accept(event) else 'skip'}")

new crash report       -> run the agent
edited crash report    -> skip
bot-opened issue       -> skip
feature request        -> skip
ping (no issue key)    -> skip

Three things in this expression are worth knowing:

• `action` is a bare name. Any name that isn't one of the five built-in names is looked up in the payload, so action means payload['action'].

• `payload.issue.user.type` is a dotted lookup. It works on dicts and is the same as payload['issue']['user']['type'].

• The ping event has no `issue` key and is skipped without an error. A missing key gives None at every level, so payload.issue.user.type is None and contains(...) is false.

Testing filters with compile_filter and sample payloads is cheap enough to do in unit tests. Save real payloads from your provider's delivery log as fixtures.

The expression is parsed with Python's ast module and checked against a whitelist when the TriggerConfig is created. Anything outside it is rejected with a ValidationError that names the problem:

"""Step 2: what fails when the config is built, and what fails quietly on an event."""
import logging

from pydantic import ValidationError

from promptise.runtime import TriggerConfig, TriggerEvent
from promptise.runtime.triggers import compile_filter

logging.basicConfig(format="%(levelname)s %(name)s: %(message)s")

print("-- rejected when the TriggerConfig is created:")
for expression in (
    "payload.get('action') == 'opened'",      # dict methods are not allowed
    "payload['labels'][0:2]",                  # slices are not allowed
    "payload.count * 2 > 10",                  # arithmetic is not allowed
    "any(l == 'bug' for l in labels)",         # comprehensions / other functions
    "payload.__class__",                       # dunder names
    "action = 'opened'",                       # a typo: assignment, not comparison
):
    try:
        TriggerConfig(type="cron", cron_expression="0 * * * *", filter_expression=expression)
    except ValidationError as exc:
        print(" ", exc.errors()[0]["msg"].removeprefix("Value error, "))

print("-- accepted, but skips the event at run time:")
event = TriggerEvent(trigger_id="github", payload={"comments": "5", "issue": {}})
for expression in (
    "payload.comments > 3",                    # "5" > 3: str vs int raises -> no match
    "payload.issue.milestone.title == 'v2'",   # missing keys are None -> False, no warning
    "int(payload.comments) > 3",               # convert first: matches
):
    print(f"  {expression!r:<42} -> {compile_filter(expression)(event)}")

-- rejected when the TriggerConfig is created:
  Invalid filter_expression "payload.get('action') == 'opened'": method 'get' is not allowed (allowed: endswith, lower, startswith, strip, upper)
  Invalid filter_expression "payload['labels'][0:2]": slices are not allowed
  Invalid filter_expression 'payload.count * 2 > 10': BinOp is not allowed
  Invalid filter_expression "any(l == 'bug' for l in labels)": function 'any' is not allowed (allowed: bool, contains, endswith, float, int, len, lower, startswith, str, upper)
  Invalid filter_expression 'payload.__class__': attribute '__class__' is not allowed
  Invalid filter_expression "action = 'opened'": invalid syntax
-- accepted, but skips the event at run time:
WARNING promptise.runtime.triggers.filters: filter_expression 'payload.comments > 3' could not be evaluated ('>' not supported between instances of 'str' and 'int'); skipping event 50fe4534-f1c1-40f2-b0dc-4f3aaa6d59c8
  'payload.comments > 3'                     -> False
  "payload.issue.milestone.title == 'v2'"    -> False
  'int(payload.comments) > 3'                -> True

The second half shows the run-time behavior. An expression that is valid but fails on a particular event (comparing the string "5" with the number 3) doesn't raise: the event counts as not matching and a warning is logged. A lookup through missing keys simply gives None. Convert JSON strings explicitly with int() or float() when you compare numbers.

What the language allows:

Not allowed: arithmetic, slices, comprehensions, lambdas, keyword arguments, dict methods such as .get(), names or attributes that start with _, and calls to anything else. An expression may be up to 2,000 characters and 200 syntax nodes. Two details catch people out: lower(None) returns the string "none", and a payload key named like a built-in name (trigger_type, metadata) is shadowed by it; use payload['trigger_type'] for those.

Some rules don't fit the language. GitHub sends labels as a list of objects ([{"name": "bug", ...}]), and the expression language has no comprehensions to search it. Pass a function instead; it receives the TriggerEvent:

"""Step 3: a Python function as the filter, for checks the expression language can't express."""
import logging

from promptise.runtime import ProcessConfig, RuntimeConfig, TriggerConfig, TriggerEvent
from promptise.runtime.triggers import compile_filter

logging.basicConfig(format="%(levelname)s %(name)s: %(message)s")


def is_new_bug(event: TriggerEvent) -> bool:
    """GitHub sends labels as a list of objects: [{"name": "bug", ...}, ...]."""
    issue = event.payload.get("issue") or {}
    labels = {label["name"] for label in issue.get("labels", [])}
    return event.payload.get("action") in ("opened", "labeled") and "bug" in labels


accept = compile_filter(is_new_bug)  # what the process does with the function
for name, payload in {
    "opened, labeled bug": {"action": "opened", "issue": {"labels": [{"name": "bug"}]}},
    "opened, no labels": {"action": "opened", "issue": {"labels": []}},
    "label without a name": {"action": "labeled", "issue": {"labels": [{"color": "f00"}]}},
}.items():
    print(f"{name:<22} -> {accept(TriggerEvent(trigger_id='github', payload=payload))}")

trigger = TriggerConfig(type="webhook", webhook_path="/github", filter_expression=is_new_bug)
try:
    RuntimeConfig(processes={"triage": ProcessConfig(triggers=[trigger])}).to_dict()
except Exception as exc:
    print("to_dict():", type(exc).__name__, "-", str(exc).splitlines()[0])

opened, labeled bug    -> True
opened, no labels      -> False
WARNING promptise.runtime.triggers.filters: filter callable raised KeyError('name'); skipping event ce40d931-edb8-4300-a0ed-761c53a1c4c7
label without a name   -> False
to_dict(): PydanticSerializationError - Unable to serialize unknown type: <class 'function'>

The function's exceptions are handled the same way as expression errors: the label without a name raised KeyError, the event was skipped and a warning logged. Write function filters defensively anyway (label.get("name")), so a logged warning means something unexpected.

Three limits of function filters:

• They must be plain functions. The runtime calls the filter and uses the truth value of what it returns. An async def function returns a coroutine, which counts as true, so every event passes.

• They can't be serialized. RuntimeConfig.to_dict() raises, and manifests and the control API need the string form.

• They change the trigger ID. A trigger's stable ID is a digest of its config. With a function filter, the digest includes the function's repr, which differs between program runs, so the ID is stable within one program but not across restarts of it.

Prefer the string form when it can express the rule.

Event and message triggers wrap the original event in their own payload. An event trigger's payload is {"event_type", "event_id", "source", "data"}, so the fields your producer sent are under data. A message trigger's payload is {"topic", "message_id", "sender", "content"}. The example below wakes an incident agent only for high and critical alerts from monitoring:

"""Step 4: filter events from an event bus; the process counts what it skips."""
import asyncio
import logging
import os
from dataclasses import dataclass, field
from uuid import uuid4

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8792/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.getLogger("promptise").setLevel(logging.ERROR)

from fake_llm import FakeLLM
from promptise.runtime import AgentRuntime, ProcessConfig, TriggerConfig


@dataclass
class BusEvent:
    data: dict
    source: str = ""
    event_id: str = field(default_factory=lambda: uuid4().hex)


class Bus:
    """The smallest event bus EventTrigger accepts."""

    def __init__(self) -> None:
        self.handlers: dict[str, list] = {}

    async def subscribe(self, event_type, handler):
        self.handlers.setdefault(event_type, []).append(handler)

    async def unsubscribe(self, event_type, handler):
        self.handlers.get(event_type, []).remove(handler)

    async def emit(self, event_type, data, source=""):
        for handler in list(self.handlers.get(event_type, [])):
            await handler(BusEvent(data=data, source=source))


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    bus = Bus()
    runtime = AgentRuntime(event_bus=bus)
    process = await runtime.add_process("incident-agent", ProcessConfig(
        model="openai:gpt-5-mini",
        instructions="Investigate production alerts.",
        triggers=[TriggerConfig(
            type="event",
            event_type="alert.raised",
            event_source="monitoring",                        # checked by the trigger
            filter_expression="payload.data.severity in ('high', 'critical')",
        )],
    ))
    await runtime.start_all()

    for severity in ["low", "high", "low", "medium", "critical", "low", "low", "high"]:
        await bus.emit("alert.raised", {"severity": severity, "service": "api"},
                       source="monitoring")
    for _ in range(3):
        await bus.emit("alert.raised", {"severity": "critical"}, source="load-test")
    await asyncio.sleep(1.0)

    status = process.status()
    print("runs:", status["invocation_count"], "| filtered_count:", status["filtered_count"],
          "| model calls:", llm.calls)
    print("prompt of the first run:\n" + llm.prompts[0])
    await runtime.stop_all()
    await llm.stop()


asyncio.run(main())

runs: 3 | filtered_count: 5 | model calls: 3
prompt of the first run:
[Trigger: event] trigger_id=event-5b15ec282b event_id=6ef81d36-b4b0-4369-ad1b-101cf9efc65c at=2026-10-20T21:21:42.558344+00:00
…
<untrusted-trigger-payload-86cae60a65f8>
{
  "event_type": "alert.raised",
  "event_id": "f4d5c59f8af94f1aae3f38c4bf68041e",
  "source": "monitoring",
  "data": {
    "severity": "high",
    "service": "api"
  }
}
</untrusted-trigger-payload-86cae60a65f8>

Eight alerts came from monitoring and three from a load test. The trigger's event_source dropped the load-test alerts before the filter saw them, so they are not in filtered_count. The filter skipped five of the eight monitoring alerts (low and medium) and counted them. Three runs, three model calls. The prompt shows the payload shape the expression reads: payload.data.severity.

Use event_source for the sender and filter_expression for the content. For the bus side of this (writing a bus, wildcards, chaining agents), see the multi-agent event bus guide.

To see what a filter is worth, send the same stream of events to an agent with and without it, and count the model calls. The script sends 40 GitHub issue deliveries (opened, edited, labeled, closed, assigned) to a webhook trigger, once without a filter and once with action == 'opened' and payload.issue.user.type != 'Bot'. The scripted model server counts every request.

"""Step 5: send the same webhook stream to an agent with and without a filter, count model calls."""
import asyncio
import logging
import os
from collections import Counter

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8792/v1"  # the scripted model server
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.getLogger("promptise").setLevel(logging.ERROR)

import aiohttp

from fake_llm import FakeLLM
from promptise.runtime import AgentProcess, ProcessConfig, TriggerConfig

FILTER = "action == 'opened' and payload.issue.user.type != 'Bot'"


def github_stream() -> list[dict]:
    """40 issue deliveries, in the mix a busy repository sends."""
    kinds = ["opened"] * 8 + ["edited"] * 12 + ["labeled"] * 10 + ["closed"] * 6 + ["assigned"] * 4
    stream = []
    for n, action in enumerate(kinds):
        bot = action == "opened" and n % 4 == 3  # two of the eight are opened by a bot
        stream.append({"action": action, "issue": {
            "number": 100 + n, "title": f"Issue {100 + n}",
            "user": {"login": "renovate[bot]" if bot else "sam", "type": "Bot" if bot else "User"},
        }})
    return stream


async def run(filter_expression: str | None, port: int, llm: FakeLLM) -> None:
    config = ProcessConfig(
        model="openai:gpt-5-mini",
        instructions="Triage new GitHub issues.",
        triggers=[TriggerConfig(type="webhook", webhook_path="/github", webhook_port=port,
                                filter_expression=filter_expression)],
    )
    process = AgentProcess("issue-triage", config)
    await process.start()
    llm.calls = 0
    answers: Counter[str] = Counter()
    async with aiohttp.ClientSession() as http:
        for delivery in github_stream():
            async with http.post(f"http://127.0.0.1:{port}/github", json=delivery) as resp:
                answers[f"{resp.status} {(await resp.json())['status']}"] += 1
    while process.status()["invocation_count"] < answers["202 accepted"]:
        await asyncio.sleep(0.05)  # wait for the accepted events to run
    status = process.status()
    await process.stop()
    label = "with filter" if filter_expression else "without filter"
    print(f"{label:<15} webhook answers={dict(answers)}  agent runs={status['invocation_count']}"
          f"  model calls={llm.calls}  filtered_count={status['filtered_count']}")


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    await run(None, 9301, llm)
    await run(FILTER, 9302, llm)
    await llm.stop()


asyncio.run(main())

without filter  webhook answers={'202 accepted': 40}  agent runs=40  model calls=40  filtered_count=0
with filter     webhook answers={'202 accepted': 6, '200 ignored': 34}  agent runs=6  model calls=6  filtered_count=0

Without the filter, all 40 deliveries were accepted and ran the agent: 40 model calls. With it, the webhook ignored 34 and accepted 6 (the six issues opened by people), and the model got 6 calls. This agent has no tools, so each run is exactly one model call. A real triage agent with tools often makes several model calls per run, so the calls saved are a multiple of the 34 skipped events.

Notice filtered_count=0 in both lines. The webhook applies the filter itself and doesn't report to the process, so for webhooks count the 200 ignored answers on the sender's side (GitHub shows them in the delivery log) or wrap a function filter with your own counter. For every other trigger type, filtered_count is the number.

A manifest keeps the filter next to the rest of the agent's definition, in version control. Manifests take the string form:

version: "1.0"
name: issue-triage
model: openai:gpt-5-mini
instructions: |
  Triage new GitHub issues: label them and ask for missing details.

triggers:
  - type: webhook
    webhook_path: /github
    webhook_port: 9090
    hmac_secret: ${GITHUB_WEBHOOK_SECRET}
    signature_scheme: github
    filter_expression: >-
      action == 'opened'
      and payload.issue.user.type != 'Bot'
      and not startswith(lower(payload.issue.title), '[wip]')

Validate it before you deploy. The CLI parses the filter with the same checks as TriggerConfig, so a typo fails here, with the trigger's position, instead of at the first webhook. The second file is the same manifest with = instead of ==:

$ promptise runtime validate issue-triage.agent
Validating issue-triage.agent...
✓ Schema validation passed
⚠ No MCP servers configured — agent has no tools
…
✓ Validation complete!
$ promptise runtime validate broken-triage.agent
Validating broken-triage.agent...
✗ Schema validation failed:
Manifest validation failed: 1 validation error for AgentManifestSchema
  Value error, triggers[0] (webhook): Value error, Invalid filter_expression "action =
'opened' and payload.issue.user.type != 'Bot' and not
startswith(lower(payload.issue.title), '[wip]')": invalid syntax [type=value_error,
input_value={'version': '1.0', 'name'...sue.title), '[wip]')"}]}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.14/v/value_error

Run it with GITHUB_WEBHOOK_SECRET set and promptise runtime start issue-triage.agent; the webhook trigger guide covers the GitHub side. The example files for each step are in runs/ai-agent-event-filter/.

Test filters like code

A filter that is too strict fails silently: the agent just never runs. Keep a few real payloads per event type as fixtures and assert the decision for each with compile_filter, as step 1 does. Run the tests in CI, and add a fixture whenever a provider changes its payload format.

Watch what you skip

Graph filtered_count (from status() or GET /api/v1/processes/{name} on the control API) next to invocation_count. A sudden jump in the skipped share often means the payload format changed, not that the world got quieter. Filter warnings ("could not be evaluated") are logged on promptise.runtime.triggers.filters; alert on them.

Change a filter without a restart

process.add_trigger() and remove_trigger() change triggers while the process runs. Editing a filter changes the trigger's config, and with it the trigger's ID: add the new trigger, then remove the old one by its ID. The control API exposes the same operations under /api/v1/processes/{name}/triggers.

Don't confuse filtering with security

Filters run on data the sender controls. Authenticate webhooks with hmac_secret, restrict allowed_sources, and scan payloads for prompt injection with TriggerDeliveryConfig(scan_payloads=True). A filter keeps costs down; it doesn't keep attackers out.

Budget the events that pass

A filter reduces runs; it doesn't cap them. A burst of matching events (a mass import of issues) still runs the agent for each. Pair the filter with run and cost limits from AI agent cost control.

• No async filters. A filter is a synchronous function or expression. Checks that need a database or an API belong in a hook or in the agent.

• No arithmetic or comprehensions in the expression language; use a function.

• Webhook skips aren't counted in the process status.

• Skipped events leave no trace beyond a debug log line and the counter. If you need an audit of what was skipped, log it in a function filter.

• Filters see one event at a time. Deduplication, rate limits per sender or "only the latest per issue" need state; keep that in your own code or a custom trigger.

How do I filter events before an LLM call?

Add a deterministic check between the trigger and the model. In the Promptise runtime, set filter_expression on the TriggerConfig, either as a safe expression string or as a Python function that receives the event. Events that don't match are skipped before they are queued, so the model never sees them and no tokens are spent.

Is filter_expression safe with untrusted payloads?

The expression is written by you, not by the sender, and it is parsed into a syntax tree and evaluated by a whitelist interpreter instead of Python's eval. Only lookups, comparisons, logic and a few string functions are allowed. The payload is data the expression reads; it can't run code. A sender can still set fields to pass the filter, so don't use a filter as a security control.

What happens when a filter hits a missing field?

A lookup through missing keys returns None, so payload.issue.milestone.title == 'v2' is simply false for an event without a milestone. If an operation raises, such as comparing None with a number, the event counts as not matching and a warning is logged. The agent keeps running.

Can I use a Python function as an event filter?

Yes. Pass a plain function that takes the TriggerEvent and returns a bool. Use it for logic the expression language can't express, such as searching a list of label objects. It must not be async, it can't be saved in a manifest or sent over the control API, and exceptions it raises skip the event.

How much does event filtering save?

Every skipped event saves a full agent run: at least one model call, often several when the agent uses tools. In this guide's example, a filter for newly opened issues cut 40 GitHub deliveries down to 6 runs. Measure your own ratio by comparing filtered_count (or the webhook's ignored answers) with invocation_count.

Should I filter with an LLM instead?

Only for rules about meaning, such as "is this a bug report?". A small model classifier costs one cheap call per event; a filter expression costs nothing. Use a filter for facts in the payload and a model for judgment, and put the filter first.

• The webhook side, with signatures and response codes: AI agent webhook trigger.

• Chain agents on events, then filter them: multi-agent event bus.

• What happens to events that pass but fail: AI agent retries and a dead letter queue.

• Overview of every trigger type: event-driven AI agents.

• Cap the runs that do pass: AI agent cost control.

• Reference: filtering events before the agent runs in the Promptise docs.