Emitters: schedules & webhooks

Start capabilities on a schedule, an authenticated webhook or a memory change. Teams declare which source invokes which capability; the server stores the trigger and admits matching work. This chapter covers payload conversion, trigger context, delivery history and replay. A running server and worker are required.

Webhook verification reads the secret named by secret_env from the graph server's environment, not the worker's. Coordinate that setting with the server operator and configure the external sender to use the generated webhook URL.

How emitters work

An emitter is a stored server-side trigger definition. Declare bindings with triggers=[...] on a capability, or the @on_schedule and @on_event decorators. With default Agent(sync_emitters=True), worker startup registers the node and calls app.sync_emitters(). Each (capability, source) binding creates or updates a code-managed emitter. Removed bindings become disabled and orphaned.

Emitters created with Emitters.create or the Console are UI-managed. Code synchronization leaves those definitions alone.

Input handling depends on how the emitter is managed (python/atfgraph/emitters.py):

• Code-managed: Admission skips the capability input-schema check. The worker applies any transform, binds the result to parameters and validates it before calling the handler.
• UI-managed: The server validates the delivered input at admission. The worker receives it unchanged, without a code-managed transform.

A capability can bind each source only once. Two GitHub EventTrigger bindings on the same capability raise ValueError when the module is imported.

Declare a schedule

Choose either cron or every_seconds in ScheduleTrigger. Cron accepts five fields, or six with leading seconds. Intervals range from 1 to 31536000 seconds. Set an IANA timezone, default UTC, and an input object delivered on each tick. The schedule example declares:

from __future__ import annotations

from atfgraph import Agent, ScheduleTrigger, TriggerContext, on_schedule

app = Agent(node_id="ticker", version="1.0.0", description="Scheduled fixture ticks")


@app.traversal(triggers=[ScheduleTrigger(cron="* * * * * *", input={"region": "AU"})])
async def tick(region: str, trigger: TriggerContext) -> dict:
    count = await app.memory.global_scope().update(f"ticks:{region}", lambda current: (current or 0) + 1)
    return {"region": region, "count": count, "delivery_key": trigger.idempotency_key, "emitter_id": trigger.emitter_id}


@app.skill()
@on_schedule(every_seconds=1)
async def heartbeat(trigger: TriggerContext | None = None) -> dict:
    beats = await app.memory.global_scope().update("heartbeats", lambda current: (current or 0) + 1)
    return {"beats": beats, "via_emitter": trigger is not None}


if __name__ == "__main__":
    app.run()

Keep @app.traversal() outermost; @on_schedule attaches the trigger declaration for it to register. Starting the worker logs the synchronized definitions:

python -m atfgraph.worker --module examples.python.features.schedule_emitter.agent:app
python -m examples.python.features.schedule_emitter.inspect_emitters

The real-daemon emitter tests check the name code:ticker:heartbeat:cron, source cron, event type tick, code management and unique schedule: delivery keys for each due time. Those keys prevent duplicate admission of the same tick after restart. This example needs no model provider.

Declare a webhook

Bind a webhook with EventTrigger(source, types, secret_env, config, transform). The server filters types by exact match or dotted prefix; an empty list or "*" accepts all. Generic sources use config for headers and timestamp tolerance. Business conditions inside a payload belong in application code. For example, GitHub's issues event type covers several actions; check event["action"] == "opened" in the handler if only newly opened issues should proceed.

Not verified: the full Databricks source header and payload layout by the examples cited here. Its documented options and test creation do not alone establish live verification. The GitHub, Stripe, Slack and generic sources have end-to-end coverage in the emitter and migrated-operations tests.

The webhook example declares these handlers:

from __future__ import annotations

from typing import Any, Dict, Optional

from pydantic import BaseModel

from atfgraph import Agent, EventTrigger, TriggerContext, on_event

app = Agent(node_id="hooks", version="1.0.0")


class Payment(BaseModel):
    id: str
    amount: int
    currency: str = "aud"
    customer: Optional[str] = None


def _payment_from_event(event: Dict[str, Any]) -> Dict[str, Any]:
    obj = event["data"]["object"]
    return {"id": obj["id"], "amount": obj["amount"], "currency": obj.get("currency", "aud"), "customer": obj.get("customer")}


@app.traversal(
    triggers=[
        EventTrigger(
            source="generic_bearer",
            types=["payment.succeeded"],
            secret_env="HOOKS_PAYMENT_SECRET",
            config={"idempotency_header": "X-Delivery-Id", "event_type_header": "X-Event-Type"},
            transform=_payment_from_event,
        )
    ]
)
async def handle_payment(payment: Payment, trigger: TriggerContext | None = None) -> dict:
    return {"payment_id": payment.id, "amount_cents": payment.amount, "delivery_key": trigger.idempotency_key if trigger else None}


@app.traversal()
@on_event("github", types=["pull_request"], secret_env="HOOKS_GITHUB_SECRET")
async def handle_pr(event: dict, trigger: TriggerContext | None = None) -> dict:
    pr = event.get("pull_request") or {}
    return {"action": event.get("action"), "title": pr.get("title")}

Set HOOKS_PAYMENT_SECRET and HOOKS_GITHUB_SECRET on the daemon, then start the worker. After synchronization, app.webhook_urls() returns a URL per bound handler, such as .../hooks/<id>. Configure the external sender with that server-owned /hooks/{emitter_id} URL.

Declare a memory trigger

Use source memory to react to writes and deletes without a polling listener. The emitter has kind event. Its source_config selects scope, optional scope ID and key prefix, entry kinds (value, vector) and actions (set, delete). A matching change submits {key, scope, scope_id, action, kind, revision, principal, value?}. Set events include value only when it is at most 16 KiB.

The delivery key memory:<scope>:<scope_id>:<key>:<revision> suppresses duplicate admission. A failed dispatch does not fail the original memory write; GET /api/v2/emitters/metrics counts it under memory_dispatch_failures.

from atfgraph import Agent
from atfgraph.decorators import on_event

app = Agent(node_id="briefs")


@app.traversal()
@on_event(source="memory", config={"scope": "global", "prefix": "brief.", "actions": ["set"]})
async def brief_changed(key: str, action: str, revision: int, value: dict | None = None) -> dict:
    return {"key": key, "revision": revision, "summary": (value or {}).get("destination")}

The equivalent Console flow is Build → Triggers → Create trigger → "memory change".

Transforms and input binding

Transforms must be synchronous; declaring an async def transform raises TypeError. For code-managed deliveries, the worker transforms the raw event, then binds it. A dict covering the handler's required data parameters binds by keyword. Otherwise, a single data parameter receives the whole payload, as event does in handle_pr. A handler with no data parameters ignores the payload. Unsupported shapes fail with invalid_input.

A transform exception, such as a missing envelope field, fails the run with transform_failed while the worker keeps serving. Valid transformation followed by invalid typed input, such as amount="lots", fails with invalid_input. Daemon tests verify that neither case invokes the handler.

Read the TriggerContext

Ask for injected context with a parameter named trigger or webhook, or a TriggerContext annotation. Give it a None default when direct calls are also supported, since those calls have no emitter context.

Within a handler, app.ctx.trigger and app.ctx.via_emitter provide the same trigger context.

Duplicate suppression, filtering and dead letters

The real-daemon emitter tests establish these delivery outcomes:

• Invalid bearer or HMAC authentication returns 401 and starts no run.
• Reusing a delivery key with the same body returns 202 with the existing event and traverser IDs. Changing the body under that key returns 409.
• An unmatched event type is recorded as filtered without a traverser.
• Invalid raw input for a UI-managed emitter returns 202 with event status failed and invalid_input. It enters the dead-letter list, counted by dlq_depth.

Manage emitters from code

Manage definitions through app.emitters, or Emitters(client) outside an agent. They wrap /api/v2/emitters. The schedule inspector and daemon tests use:

import asyncio
import os

from atfgraph import Client, Emitters


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        emitters = Emitters(client)
        for emitter in await emitters.list():
            print(emitter.name, emitter.target, emitter.managed_by, emitter.enabled, emitter.orphaned)
            for event in (await emitters.events(emitter.id))[:5]:
                print("   ", event.delivery_key, event.status, event.traverser_id)
        manual = await emitters.create(
            "manual-receive", "hooks.receive", source="generic_bearer",
            secret_env="HOOKS_PAYMENT_SECRET", source_config={"idempotency_header": "X-Delivery-Id"},
        )
        fired = await emitters.fire(manual.id, "fire-1", input={"value": 7})
        print(await fired.result(timeout=10))
        current = await emitters.get(manual.id)
        await emitters.update(manual.id, expected_revision=current.revision, enabled=False)


if __name__ == "__main__":
    asyncio.run(main())

fire manually submits validated input with an explicit delivery key. Reusing the key returns the same traverser, and no trigger context is attached. replay(emitter_id, event_id) uses a new key to execute a stored inbound event again.

Updates require expected_revision. Disabling a code-managed emitter sets sticky_paused, and its webhook returns 404 until re-enabled. It cannot be deleted while its code binding exists: that returns 409. Remove the declaration and synchronize to orphan it. Converting it with managed_by="ui" transfers management to the UI; the reverse conversion is unavailable. Creating, synchronizing, replaying and updating emitters require administrator authority.

Test a handler without a server

simulate_trigger invokes a handler locally with a synthetic TriggerContext. For code-managed bindings, it performs the same transformation, input binding and validation as the worker. The migrated-operations test uses:

import asyncio

from atfgraph import simulate_trigger
from examples.emitters.python.agent import app


async def main() -> None:
    result = await simulate_trigger(
        app.entry("handle_tick"), source="cron", body={}, event_type="tick",
        event_id="evt-1", idempotency_key="schedule:2026-01-01T00:00:00Z",
    )
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

Simulation still needs a server if the handler calls services such as app.memory. The shown test binds the worker to a real daemon for that reason. Use the daemon fixture to test signature verification and full delivery behavior.

Next

• Coordinate triggered work with Multi-node systems.
• Inspect its usage and logs in Observability.
• Run isolated webhook tests with Testing.