Trigger agents from GitHub, Stripe, Slack, or any custom webhook

from agentfield import Agent, EventTrigger, TriggerContext, on_event, on_schedule

app = Agent(node_id="ops-agent")

# GitHub: fire on pull_request events
@app.reasoner()
@on_event(
    source="github",
    types=["pull_request"],
    secret_env="GITHUB_WEBHOOK_SECRET",  # secret lives on the control plane
)
async def on_pr(event: dict, trigger: TriggerContext | None = None):
    pr = event["pull_request"]
    return {
        "action": event["action"],
        "title": pr["title"],
        "delivery_id": trigger.idempotency_key if trigger else None,
    }

# Stripe: fire on successful payments, with a transform that flattens the envelope
def _flatten(evt: dict) -> dict:
    obj = evt.get("data", {}).get("object", {})
    return {"id": obj.get("id"), "amount": obj.get("amount"), "customer": obj.get("customer")}

@app.reasoner(triggers=[
    EventTrigger(
        source="stripe",
        types=["payment_intent.succeeded"],
        secret_env="STRIPE_WEBHOOK_SECRET",
        transform=_flatten,
    ),
])
async def on_payment(payment: dict, trigger: TriggerContext | None = None):
    return {"stripe_id": payment["id"], "amount": payment["amount"]}

# Custom webhook with HMAC signing
@app.reasoner()
@on_event(source="generic_hmac", secret_env="MY_WEBHOOK_SECRET")
async def on_custom(payload: dict, trigger: TriggerContext | None = None):
    return {"event_type": trigger.event_type, "payload": payload}

# Cron: fire every five minutes
@app.reasoner()
@on_schedule("*/5 * * * *")
async def heartbeat(_input, trigger: TriggerContext | None = None):
    return {"fired_at": trigger.received_at.isoformat()}

app.run()