Custom AI Agent Triggers: Wake an Agent From Any Source

Write a class with three async methods: start(), stop() and wait_for_next(), which blocks until the source has an item and returns it as a TriggerEvent. Register a factory for it with register_trigger_type("sqlite_queue", factory), and configure it like any built-in trigger: TriggerConfig(type="sqlite_queue", custom_config={...}). For a queue, acknowledge each item when its run ends, with a task_completed hook, so a crash returns the item to the queue instead of losing it.

• The protocol is small: a writable trigger_id and three methods. Everything else is optional.

• `custom_config` carries your settings. Validate it in the factory, which runs when the process starts.

• Acknowledge after the run, not on receipt. That gives at-least-once delivery, so the agent's work must be idempotent.

• `drain_pending()` and `set_availability_check()` make the trigger behave well on stops and failures.

• Register before you load. A manifest can use the type, but the program that loads it must import the registration first.

An agent process runs a listener task for each trigger. The listener calls wait_for_next() in a loop and passes every event on: through the trigger's filter and the task_created hooks, into the process queue, to a worker that runs the agent. The trigger never runs the agent itself. Its only job is to turn items from a source into events.

source (job table, SQS, Redis stream)
   │  claim item (hidden from other consumers for visibility_timeout)
   ▼
trigger.wait_for_next() ──▶ listener ──▶ filter ──▶ task_created hooks ──▶ process queue
                                                                              │
                                                                         worker: run agent
                                                                              │
                                              task_completed hook (outcome) ◀─┘
                                                  │ success ──▶ mark item done
                                                  │ failed / dead-lettered ──▶ mark item failed
                                                  └ no ack (crash) ──▶ item visible again

The members the runtime uses:

drain_pending() -> list[TriggerEvent]

A TriggerEvent has trigger_id, trigger_type, event_id (a UUID by default), timestamp, payload and metadata. The agent sees the payload, inside a delimited untrusted-data block, but not the metadata. Put anything the agent needs, such as a job ID, into the payload.

Delivery semantics come from when you acknowledge. Deleting an item when you claim it is at-most-once: a crash mid-run loses it. Acknowledging after the run is at-least-once: a crash means the item runs again later. For agents, at-least-once plus idempotent tools is almost always the right choice.

Write one for pull-based sources that have no way to call you:

• Job tables in Postgres, MySQL or SQLite.

• Cloud queues: SQS, Google Pub/Sub pull subscriptions, Azure Service Bus.

• Redis Streams or Kafka consumer groups.

• Mailboxes (IMAP), ticket systems or APIs you poll.

You don't need one when the source can push. If it can send an HTTP request, use the webhook trigger. If your code already publishes events in the same program, wrap the publisher as an event bus and use an event or message trigger (multi-agent event bus). For a fixed schedule, use cron.

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

• SQLite, which ships with Python. No server to install.

• No API key: the examples use a small scripted, OpenAI-compatible model server (fake_llm.py). Point OPENAI_BASE_URL at a real provider to use a real model.

• About 25 minutes.

The module below is the whole trigger: a job table, an enqueue() function any program can call, the trigger class, its factory and an acknowledgement hook.

"""A custom trigger: wake an agent for each job in a SQLite table (a small durable work queue)."""
from __future__ import annotations

import asyncio
import json
import sqlite3
import time
from collections import deque
from contextlib import closing
from typing import Any
from uuid import uuid4

from promptise.runtime import TriggerConfig, TriggerError, TriggerEvent
from promptise.runtime.triggers import register_trigger_type

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    queue TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'ready',  -- ready | claimed | done | failed
    attempts INTEGER NOT NULL DEFAULT 0,
    claimed_at REAL,
    claimed_by TEXT,
    event_id TEXT,
    error TEXT
)"""


def connect(db_path: str) -> sqlite3.Connection:
    db = sqlite3.connect(db_path, isolation_level=None, timeout=10)  # autocommit
    db.execute(SCHEMA)
    return db


def enqueue(db_path: str, queue: str, payload: dict[str, Any]) -> int:
    """Add a job. Call it from any program that can open the database file."""
    with closing(connect(db_path)) as db:
        cursor = db.execute("INSERT INTO jobs (queue, payload) VALUES (?, ?)",
                            (queue, json.dumps(payload)))
        return int(cursor.lastrowid)


class SQLiteQueueTrigger:
    """BaseTrigger implementation: start(), stop(), wait_for_next(), plus drain_pending()."""

    trigger_type = "sqlite_queue"  # shown as "type" in process.status()["triggers"]

    def __init__(self, db_path: str, queue: str = "default", *, poll_interval: float = 1.0,
                 max_in_flight: int = 5, visibility_timeout: float = 300.0) -> None:
        self.trigger_id = f"sqlite_queue-{queue}"  # the process replaces it with a stable ID
        self.db_path = db_path
        self.queue = queue
        self.poll_interval = poll_interval
        self.max_in_flight = max_in_flight  # claimed and not yet acknowledged
        self.visibility_timeout = visibility_timeout
        self._buffer: deque[TriggerEvent] = deque()  # claimed, not yet handed over
        self._unavailable = lambda: None  # replaced by the process (set_availability_check)
        self._worker = uuid4().hex  # marks the jobs this trigger claimed

    async def start(self) -> None:
        await asyncio.to_thread(lambda: connect(self.db_path).close())  # fail early on a bad path

    async def stop(self) -> None:
        pass  # nothing runs in the background; the process cancels wait_for_next()

    def set_availability_check(self, check) -> None:
        """The process passes a callback: None while it takes work, else a reason."""
        self._unavailable = check

    async def wait_for_next(self) -> TriggerEvent:
        while not self._buffer:
            if self._unavailable() is None:  # don't claim jobs a failed process can't run
                self._buffer.extend(await asyncio.to_thread(self._claim_batch))
            if not self._buffer:
                await asyncio.sleep(self.poll_interval)
        return self._buffer.popleft()

    def drain_pending(self) -> list[TriggerEvent]:
        """Claimed jobs the process hasn't taken yet: requeued on a restart, else dead-lettered."""
        events = list(self._buffer)
        self._buffer.clear()
        return events

    @property
    def pending_events(self) -> int:  # reported in process.status()["triggers"]
        return len(self._buffer)

    def _claim_batch(self) -> list[TriggerEvent]:
        now = time.time()
        expired = now - self.visibility_timeout
        with closing(connect(self.db_path)) as db:
            db.execute("BEGIN IMMEDIATE")  # one claimer at a time, across processes
            (in_flight,) = db.execute(
                "SELECT count(*) FROM jobs WHERE queue = ? AND status = 'claimed'"
                " AND claimed_by = ? AND claimed_at >= ?", (self.queue, self._worker, expired),
            ).fetchone()
            rows = db.execute(
                "SELECT id, payload, attempts FROM jobs WHERE queue = ? AND (status = 'ready'"
                " OR (status = 'claimed' AND claimed_at < ?)) ORDER BY id LIMIT ?",
                (self.queue, expired, max(0, self.max_in_flight - in_flight)),
            ).fetchall()
            events = []
            for job_id, payload, attempts in rows:
                event = TriggerEvent(
                    trigger_id=self.trigger_id,
                    trigger_type=self.trigger_type,
                    payload={"job_id": job_id, **json.loads(payload)},  # the agent sees the id
                    metadata={"job_id": job_id, "attempt": attempts + 1},
                )
                db.execute("UPDATE jobs SET status = 'claimed', claimed_at = ?, claimed_by = ?,"
                           " attempts = attempts + 1, event_id = ? WHERE id = ?",
                           (now, self._worker, event.event_id, job_id))
                events.append(event)
            db.execute("COMMIT")
        return events


def sqlite_queue_factory(config: TriggerConfig, *, event_bus=None, broker=None):
    """Build the trigger from TriggerConfig.custom_config (called when the process starts)."""
    options = dict(config.custom_config)
    allowed = {"db_path", "queue", "poll_interval", "max_in_flight", "visibility_timeout"}
    if unknown := sorted(set(options) - allowed):
        raise TriggerError(f"sqlite_queue: unknown custom_config keys {unknown}")
    if "db_path" not in options:
        raise TriggerError("sqlite_queue: custom_config needs 'db_path'")
    return SQLiteQueueTrigger(
        options["db_path"],
        options.get("queue", "default"),
        poll_interval=float(options.get("poll_interval", 1.0)),
        max_in_flight=int(options.get("max_in_flight", 5)),
        visibility_timeout=float(options.get("visibility_timeout", 300.0)),
    )


def ack_hook(db_path: str):
    """A task_completed hook that marks each job done or failed when its run ends."""

    async def ack(ctx) -> None:
        outcome = ctx.data["outcome"]
        if ctx.data["trigger_type"] != "sqlite_queue" or outcome == "retrying":
            return  # a retry keeps the claim
        status = "done" if outcome == "success" else "failed"

        def write() -> None:
            with closing(connect(db_path)) as db:
                db.execute("UPDATE jobs SET status = ?, error = ? WHERE event_id = ?",
                           (status, ctx.data["error"], ctx.data["event_id"]))

        await asyncio.to_thread(write)

    return ack


register_trigger_type("sqlite_queue", sqlite_queue_factory, overwrite=True)

How the parts fit:

• Claiming happens in _claim_batch(), inside BEGIN IMMEDIATE, so two workers (in different programs) never claim the same job. A claim sets status='claimed', the time, this trigger's worker ID and the event's ID.

• `max_in_flight` limits how many jobs this trigger has claimed and not yet acknowledged. The listener moves events into the process queue as fast as wait_for_next() returns them, and that queue holds 1,000 events. Without a limit, a busy table would be claimed into memory all at once.

• `visibility_timeout` makes a claimed job claimable again once it is older than that many seconds. That is the crash recovery: a worker that died never acknowledges, and another one picks the job up.

• Blocking calls run in a thread (asyncio.to_thread). One process runs every agent on the same event loop, so a synchronous database call in wait_for_next() would stall all of them.

• `stop()` has nothing to do. The process cancels the listener, and with it the pending wait_for_next(), before it calls stop().

register_trigger_type("sqlite_queue", sqlite_queue_factory) runs when the module is imported. The factory receives the whole TriggerConfig (plus the runtime's event_bus and broker, unused here) and returns a trigger. TriggerConfig checks only built-in types, so the factory is where custom_config gets validated. A typo in a key shows up when the process starts:

registered: ['cron', 'event', 'file_watch', 'message', 'sqlite_queue', 'webhook']
start() raised TriggerError: sqlite_queue: unknown custom_config keys ['db']
state: failed
watch_path on a custom type: accepted, and ignored

The config said {"db": "jobs.db"} instead of db_path. start() raised the factory's TriggerError and the process ended in FAILED, before any job was claimed. The last line shows the other side: a field of a built-in type (watch_path) on a custom type passes validation and is ignored, because custom types receive the config as is. Read only custom_config in your factory and reject unknown keys there.

Registration is process-wide. register_trigger_type raises ValueError if the name is taken (pass overwrite=True to replace it, as the module does so it can be re-imported), unregister_trigger_type(name) removes it, and registered_trigger_types() lists every type.

The worker registers the type by importing the module, adds three tickets to the table, and starts a process with the custom trigger and the acknowledgement hook. The scripted model fails every call that mentions ticket T-102, to show a failed job.

"""Steps 3-4: run an agent on SQLite jobs and acknowledge each job when its run ends."""
import asyncio
import logging
import os
import sqlite3
import tempfile

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

import sqlite_queue  # registers the "sqlite_queue" trigger type
from fake_llm import FakeLLM
from promptise.runtime import AgentProcess, HookConfig, ProcessConfig, TriggerConfig

DB = os.path.join(tempfile.mkdtemp(), "jobs.db")

config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="Summarize each support ticket in one sentence for the on-call engineer.",
    triggers=[TriggerConfig(
        type="sqlite_queue",
        custom_config={"db_path": DB, "queue": "tickets", "poll_interval": 0.5},
    )],
    hooks=[HookConfig(event="task_completed", callback=sqlite_queue.ack_hook(DB))],
)


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    llm.poison = {"T-102"}  # this ticket makes the model call fail

    # Any program can add jobs, before or while the agent runs.
    for ticket in ("T-101", "T-102", "T-103"):
        sqlite_queue.enqueue(DB, "tickets", {"ticket": ticket, "text": f"Customer report {ticket}"})

    process = AgentProcess("ticket-worker", config)
    await process.start()
    print("triggers:", process.status()["triggers"])
    await asyncio.sleep(2.0)

    first_prompt = [line for line in llm.prompts[0].splitlines()
                    if not line.startswith("The payload inside")]  # skip the boilerplate line
    print("first prompt:", *first_prompt, sep="\n  ")
    rows = sqlite3.connect(DB).execute(
        "SELECT id, json_extract(payload, '$.ticket'), status, attempts, substr(error, 1, 40) "
        "FROM jobs").fetchall()
    for row in rows:
        print("job", row)
    status = process.status()
    print("runs:", status["invocation_count"], "| dead letters:", status["dead_letter_count"])
    await process.stop()
    await llm.stop()


asyncio.run(main())

triggers: [{'id': 'sqlite_queue-be362898aa', 'type': 'sqlite_queue', 'source': 'config', 'pending_events': 0}]
first prompt:
  [Trigger: sqlite_queue] trigger_id=sqlite_queue-be362898aa event_id=d51352b9-8398-4cc1-a218-103107f5180c at=2026-10-20T21:25:25.417969+00:00
  <untrusted-trigger-payload-e828b4ae71fa>
  {
    "job_id": 1,
    "ticket": "T-101",
    "text": "Customer report T-101"
  }
  </untrusted-trigger-payload-e828b4ae71fa>
job (1, 'T-101', 'done', 1, None)
job (2, 'T-102', 'failed', 1, "GraphExecutionError: Graph 'react' faile")
job (3, 'T-103', 'done', 1, None)
runs: 2 | dead letters: 1

The trigger got a stable ID (sqlite_queue-<digest>) and its pending_events counter shows up in the status. The agent's prompt carries the job as an untrusted payload, with the job_id the trigger added. Two tickets ran and were marked done; T-102's run failed, the process dead-lettered it with retries exhausted, and the hook marked the row failed with the error. The table is now the record of what happened, and failed rows are your durable dead letter queue: query them, fix the cause, set them back to ready.

ack_hook() returns a task_completed lifecycle hook. The process fires it when a queued event finishes, with event_id, trigger_id, trigger_type, outcome, duration_ms and error:

The hook finds the row by event_id, which the trigger stored when it claimed the job. Events dead-lettered without a run (at a stop, or while the process is FAILED) don't fire task_completed; their jobs stay claimed until step 5 releases them or the visibility timeout expires.

Test the crash path without a process or a model. Claim a job, don't acknowledge it, and wait out a 1-second visibility timeout:

"""Production: test the trigger on its own, without a process or a model (e.g. in CI)."""
import asyncio
import os
import tempfile

from sqlite_queue import SQLiteQueueTrigger, enqueue

DB = os.path.join(tempfile.mkdtemp(), "jobs.db")


async def main() -> None:
    enqueue(DB, "tickets", {"ticket": "T-401"})
    trigger = SQLiteQueueTrigger(DB, "tickets", poll_interval=0.1, visibility_timeout=1.0)
    await trigger.start()

    first = await trigger.wait_for_next()
    print("claimed:", first.payload, first.metadata)
    try:  # nothing is ready: the claim is still visible to us only
        await asyncio.wait_for(trigger.wait_for_next(), timeout=0.5)
    except asyncio.TimeoutError:
        print("no second claim while the first one is fresh")

    await asyncio.sleep(1.0)  # the "worker" crashed: no ack within visibility_timeout
    again = await asyncio.wait_for(trigger.wait_for_next(), timeout=1.0)
    print("claimed again:", again.payload, again.metadata,
          "| same event_id:", again.event_id == first.event_id)
    await trigger.stop()


asyncio.run(main())

claimed: {'job_id': 1, 'ticket': 'T-401'} {'job_id': 1, 'attempt': 1}
no second claim while the first one is fresh
claimed again: {'job_id': 1, 'ticket': 'T-401'} {'job_id': 1, 'attempt': 2} | same event_id: False

The job came back with attempt: 2 and a new event_id. That is why the trigger puts job_id in the payload: it is the stable key for idempotent tools, since the event ID changes with every claim. Retries and redeliver_dead_letters() keep the event ID; a re-claim doesn't.

When a process stops, the runtime collects every event that didn't run: those in the process queue, the one a worker was running (cancelled), and whatever the trigger still holds, which it gets by calling drain_pending() after stop(). On a stop they become dead letters with reason process stopped; on an automatic restart they go back to the process queue. Without drain_pending(), events buffered inside a trigger would be lost at a stop.

For a durable queue, the better home for unrun jobs is the queue itself. After stop(), give them back:

def release_unfinished(process: AgentProcess, db_path: str) -> tuple[int, list[dict]]:
    """After stop(): put jobs that never finished back to 'ready'; return the other letters."""
    released, others = 0, []
    with closing(sqlite_queue.connect(db_path)) as db:
        for letter in process.dead_letters:
            event = letter["event"]
            if event.trigger_type == "sqlite_queue" and letter["reason"] == "process stopped":
                db.execute("UPDATE jobs SET status = 'ready', claimed_at = NULL "
                           "WHERE event_id = ? AND status = 'claimed'", (event.event_id,))
                released += 1
            else:
                others.append(letter)
    process.clear_dead_letters()  # the queue owns the released jobs again
    return released, others

clean_stop.py adds five tickets, makes the model slow, and stops the process half a second after it starts:

after stop: [('process stopped', 'run cancelled'), ('process stopped', None)] ...
jobs: [(1, 'claimed', 1), (2, 'claimed', 1), (3, 'claimed', 1), (4, 'claimed', 1), (5, 'claimed', 1)]
released: 5 | other dead letters: 0
jobs: [(1, 'done', 2), (2, 'done', 2), (3, 'done', 2), (4, 'done', 2), (5, 'done', 2)]

All five jobs were claimed (the limit is 5). The stop cancelled the run in flight and dead-lettered the rest. release_unfinished put the five rows back to ready, and the next process ran all of them. The attempt count of 2 shows that each job was claimed twice, which is fine as long as the work is idempotent.

A custom type works in a .agent manifest like a built-in one. The manifest stores the settings; the code stays in your module:

version: "1.0"
name: ticket-worker
model: openai:gpt-5-mini
instructions: |
  Summarize each support ticket in one sentence for the on-call engineer.

triggers:
  - type: sqlite_queue          # a custom type: register it before loading this file
    custom_config:
      db_path: ./jobs.db        # relative to the working directory
      queue: tickets
      poll_interval: 1.0
      max_in_flight: 5
      visibility_timeout: 300

config:
  max_consecutive_failures: 5

There is one rule: the program that loads the manifest must register the type first. promptise runtime validate ticket-worker.agent passes, because the schema accepts any type name, but promptise runtime start ticket-worker.agent fails with Unknown trigger type: 'sqlite_queue', since the CLI process never imported your module. Start such agents from Python: import the module, then AgentRuntime.load_manifest(). Manifest hooks are commands only, so register the Python acknowledgement hook on process.hooks after loading.

manifest_demo.py loads the manifest, adds the acknowledgement hook, runs two jobs, and lists the triggers over the control API. A second process with credentials in its custom_config shows what the listing masks:

"""Step 6: load a manifest that uses the custom type, run it, and list its triggers over the API."""
import asyncio
import logging
import os
import secrets
import sqlite3
import tempfile

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

import sqlite_queue  # 1. register the type in the program that loads the manifest
from fake_llm import FakeLLM
from promptise.runtime import (
    AgentRuntime, HookEvent, OrchestrationAPI, OrchestrationClient, ProcessConfig, TriggerConfig,
)

MANIFEST = os.path.abspath("ticket-worker.agent")
os.chdir(tempfile.mkdtemp())  # ./jobs.db in the manifest is relative to the working directory


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    runtime = AgentRuntime()
    process = await runtime.load_manifest(MANIFEST)  # 2. then load the manifest
    db_path = process.config.triggers[0].custom_config["db_path"]
    process.hooks.on(HookEvent.TASK_COMPLETED)(sqlite_queue.ack_hook(db_path))  # 3. ack hook

    for ticket in ("T-301", "T-302"):
        sqlite_queue.enqueue(db_path, "tickets", {"ticket": ticket})
    await runtime.start_all()
    await asyncio.sleep(2.5)
    print("jobs:", sqlite3.connect(db_path).execute("SELECT id, status FROM jobs").fetchall())

    # What GET /triggers shows, also for a trigger with credentials in custom_config.
    await runtime.add_process("stream-reader", ProcessConfig(triggers=[TriggerConfig(
        type="redis_stream",
        custom_config={"url": "redis://:s3cret@cache:6379/0", "password": "s3cret",
                       "stream": "tickets"},
    )]))
    token = secrets.token_urlsafe(16)
    api = OrchestrationAPI(runtime, port=9311, auth_token=token)
    await api.start()
    async with OrchestrationClient("http://127.0.0.1:9311", token=token) as client:
        for name in ("ticket-worker", "stream-reader"):
            for trigger in await client.list_triggers(name):
                print(name, trigger["id"], trigger["config"]["custom_config"])
    await api.stop()
    await runtime.stop_all()
    await llm.stop()


asyncio.run(main())

jobs: [(1, 'done'), (2, 'done')]
ticket-worker sqlite_queue-9a0ffe880a {'db_path': './jobs.db', 'queue': 'tickets', 'poll_interval': 1.0, 'max_in_flight': 5, 'visibility_timeout': 300}
stream-reader redis_stream-9c58561083 {'url': 'redis://:s3cret@cache:6379/0', 'password': '**********', 'stream': 'tickets'}

The other files in runs/custom-ai-agent-trigger/ are the steps above: worker.py (step 3, in Python instead of a manifest), test_trigger.py (step 4), clean_stop.py (step 5) and fake_llm.py, the scripted model server.

Keep credentials out of custom_config

GET /api/v1/processes/{name}/triggers masks custom_config values whose key looks like a credential (names containing secret, token, password, api_key, auth, credential, signature or cookie). The output above shows the limit: password was masked, but the same password inside url was not. Better still, keep secrets out of the config: put the name of an environment variable in custom_config ("password_env": "REDIS_PASSWORD") and read it in the factory.

Don't claim what a failed process can't run

Through set_availability_check(), the trigger stops claiming while the process can't take work. Together with max_in_flight, a failure leaves the backlog in the queue. In this run every job failed and the process reached max_consecutive_failures=2:

state: failed
jobs by status: [('failed', 2), ('ready', 8)]
dead letters: ['retries exhausted', 'retries exhausted']

Two jobs were claimed and failed; the other eight are still ready for the next deploy, instead of being claimed, dead-lettered with process failed and stuck until their visibility timeout. Without the check, a FAILED process keeps calling wait_for_next() and would dead-letter every job the trigger hands over. See AI agent retries and a dead letter queue for how failures, retries and restarts interact.

Choose the polling numbers

poll_interval is the delay between empty polls, so it is the worst-case wait for a new job and the main cost of an idle trigger: one small query per interval. One to five seconds suits most work queues; a job that must start within a second needs a push source instead (a webhook, or Postgres LISTEN/NOTIFY feeding an event bus). max_in_flight should cover what the process can run at once (concurrency) plus one or two jobs of look-ahead. Higher values only move jobs from the durable table into memory, where a stop turns them into dead letters. visibility_timeout must be longer than your slowest run, including retries and their backoff, or a job still running is claimed a second time. With several worker programs on one SQLite file, enable write-ahead logging once (PRAGMA journal_mode=WAL) so readers don't wait for the claiming writer.

Make the agent's work idempotent

At-least-once delivery means a job can run twice: after a crash, a stop, a visibility timeout that was too short, or a retry. Key side effects on the job_id in the payload, and set visibility_timeout well above your longest run (with retries and backoff included).

Scale with care

SQLite with BEGIN IMMEDIATE is safe for a few worker programs on one machine. For more, use a server queue with the same pattern: Postgres SELECT ... FOR UPDATE SKIP LOCKED, SQS (ReceiveMessage with a visibility timeout, DeleteMessage as the ack), or Redis Streams consumer groups (XREADGROUP, XACK, XAUTOCLAIM for stuck entries). The trigger's shape stays the same; only _claim_batch() and the hook change.

Deploy it from Python

Because the CLI can't import your registration, run the runtime from a small main.py: import the trigger module, load the manifests, and call await runtime.run_forever(), which starts every process and shuts down cleanly on SIGTERM. Deploy AI agents with Docker shows the container side.

• No acknowledgement API in the protocol. The trigger isn't told when a run ends; the task_completed hook is the way to close the loop, and it doesn't fire for events dead-lettered without a run.

• The CLI can't load custom types. Manifests that use them must be started from Python.

• `TriggerConfig` doesn't validate custom types. Your factory has to, and errors appear at start, not when the config is built.

• One event loop. Every trigger and agent in a runtime shares it, so a misbehaving trigger affects all processes.

• The trigger ID depends on the whole config. Changing any custom_config value (a rotated token included) gives the trigger a new ID.

How do I trigger an AI agent from a message queue?

Write a custom trigger whose wait_for_next() receives one message and returns it as a TriggerEvent, register it with register_trigger_type, and add it to the agent's triggers with its settings in custom_config. Acknowledge the message in a task_completed hook when the run succeeds, so a crash makes the queue deliver it again.

What methods does a custom trigger need?

A writable trigger_id attribute and three async methods: start(), stop() and wait_for_next(), which returns the next TriggerEvent. Optional members improve behavior: drain_pending() returns buffered events at a stop, set_availability_check() lets the trigger pause while the process can't take work, and pending_events shows up in the process status.

How do I acknowledge a job only after the agent finished?

Register a task_completed lifecycle hook. It receives the event ID and the outcome (success, retrying, failed, dead_lettered or blocked). Store the event ID with the job when you claim it, then mark the job done on success, leave it claimed on retrying, and mark it failed otherwise.

Can I use a custom trigger in a YAML manifest?

Yes. Set type to your registered name and put the settings under custom_config. The program that loads the manifest must import your registration first, so start it from Python with AgentRuntime.load_manifest(). The promptise runtime start command can't load custom types.

What happens to queued jobs when the agent stops?

The runtime dead-letters every event that didn't run, including those still buffered in the trigger (through drain_pending()), with reason process stopped. For a durable queue, put those jobs back to ready after stop(), or let the visibility timeout return them. On an automatic restart, the events stay queued and run after it.

Should I use Redis, SQS or a database table?

Use what you already run. A database table is enough for modest volumes and gives you a queryable history. SQS and Redis Streams add visibility timeouts and acknowledgements natively and scale further. The trigger looks the same for each; only the claim and acknowledgement calls change.

• Overview of the built-in triggers: event-driven AI agents.

• Retries, dead letters and idempotency: AI agent retries and a dead letter queue.

• Skip events before they run: AI agent event filters.

• The task_created and task_completed hooks: AI agent lifecycle hooks.

• Declaring agents in YAML: AI agent manifest.

• Reference: custom trigger types in the Promptise docs.