← All code samples

Webhook + schedule + HITL (complete)

Customer-style automation loop using the expanded public Aiden surface:

  1. Create a webhook and a schedule against an existing agent
  2. Trigger the schedule (/trigger) and the webhook (one-time token)
  3. Resolve pending HITL approvals
  4. Wait on subscribeExecution (SSE) and list recent tool calls
  5. Delete the demo webhook and schedule

This is the same shape appcd-e2e uses for unattended webhook/schedule suites, but expressed through stackgen-sdk so customer code and CI share one client.

How to run

export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>

pip install stackgen-sdk==0.2.0
python examples/automation/webhook-schedule-hitl.py --agent my-oncall-agent

Minimal

from stackgen import StackgenClient, StackgenConfig

client = StackgenClient(
    StackgenConfig(
        base_url="https://app.stackgen.com",
        api_token="stackgen_…",
        workspace_id="<workspace-uuid>",
    )
)

wh = client.aiden.create_webhook({
    "name": "ops-pager",
    "target_type": "agent",
    "target_name": "my-oncall-agent",
    "action": "Investigate the alert",
    "enabled": True,
})
sched = client.aiden.create_schedule({
    "name": "ops-hourly",
    "target_type": "agent",
    "target_name": "my-oncall-agent",
    "expression": "0 * * * *",
    "action": "Hourly health check",
    "enabled": True,
})
client.aiden.trigger_schedule(sched["id"])

# After a run starts:
# client.aiden.wait_for_subscribe_complete(trace_id)
# client.aiden.resolve_approval(approval_id, {"decision": "approve"})

Complete scripts

#!/usr/bin/env python3
"""Real-world Aiden automation: webhook + schedule + HITL resolve + SSE wait.

Creates a short-lived webhook and a never-ticking schedule against an existing
agent, triggers both paths, resolves any pending approvals, and waits on the
execution SSE stream until complete.

Requires STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID, and an agent name.

Usage:
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_WORKSPACE_ID=<workspace-uuid>
  python examples/automation/webhook-schedule-hitl.py --agent my-oncall-agent
  # Empty CI workspaces: prefer a published workflow (agents need an integration).
  python examples/automation/webhook-schedule-hitl.py --workflow my-note-workflow
"""

from __future__ import annotations

import argparse
import os
import sys
import uuid

from stackgen import StackgenClient, StackgenConfig


def _approval_for_run(item: dict, session_id: str, trace_id: str) -> bool:
    if not isinstance(item, dict):
        return False
    keys = (
        "session_id",
        "sessionId",
        "trace_id",
        "traceId",
        "execution_id",
        "executionId",
    )
    values: set[str] = set()

    def collect(obj: dict) -> None:
        for key in keys:
            value = str(obj.get(key) or "").strip()
            if value:
                values.add(value)

    collect(item)
    for nested_key in ("execution", "run", "session"):
        nested = item.get(nested_key)
        if isinstance(nested, dict):
            collect(nested)
    if session_id and session_id in values:
        return True
    return bool(trace_id and trace_id in values)


def env(name: str, *fallbacks: str) -> str:
    for key in (name, *fallbacks):
        value = (os.environ.get(key) or "").strip()
        if value:
            return value
    raise SystemExit(f"missing env {name}")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    target = parser.add_mutually_exclusive_group(required=True)
    target.add_argument("--agent", help="Existing agent name to target")
    target.add_argument("--workflow", help="Existing workflow name to target")
    parser.add_argument(
        "--message",
        default="e2e: investigate checkout latency spike",
        help="Payload / schedule action text",
    )
    parser.add_argument("--timeout-seconds", type=int, default=600)
    args = parser.parse_args()
    target_type = "agent" if args.agent else "workflow"
    target_name = args.agent or args.workflow

    client = StackgenClient(
        StackgenConfig(
            base_url=env("STACKGEN_URL"),
            api_token=env("STACKGEN_TOKEN"),
            workspace_id=env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
            timeout_seconds=args.timeout_seconds,
        )
    )
    suffix = uuid.uuid4().hex[:8]
    webhook_id = ""
    schedule_id = ""
    trace_id = ""
    session_id = ""

    try:
        # 1) Create webhook (token only returned on create)
        wh = client.aiden.create_webhook(
            {
                "name": f"sdk-demo-wh-{suffix}",
                "target_type": target_type,
                "target_name": target_name,
                "action": args.message,
                "enabled": True,
            }
        )
        webhook_id = str(wh.get("id") or "")
        token = str(wh.get("token") or "")
        print(f"created webhook id={webhook_id}")

        # 2) Create a schedule that never ticks on its own; we /trigger it
        sched = client.aiden.create_schedule(
            {
                "name": f"sdk-demo-sched-{suffix}",
                "target_type": target_type,
                "target_name": target_name,
                "expression": "0 0 1 1 *",
                "action": args.message,
                "enabled": True,
            }
        )
        schedule_id = str(sched.get("id") or "")
        print(f"created schedule id={schedule_id}")

        # 3) Trigger schedule (same InternalTask path as cron)
        trig = client.aiden.trigger_schedule(schedule_id)
        print(f"schedule trigger: {trig}")

        # 4) Trigger webhook with the one-time token
        if token:
            client_wh = StackgenClient(
                StackgenConfig(
                    base_url=env("STACKGEN_URL"),
                    api_token=env("STACKGEN_TOKEN"),
                    workspace_id=env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
                    webhook_token=token,
                    timeout_seconds=args.timeout_seconds,
                )
            )
            started = client_wh.aiden.trigger_webhook(args.message)
            print(f"webhook trigger: {started}")
            webhook_id = str(started.get("webhook_id") or webhook_id)
            invocation = str(started.get("invocation_id") or "")
            if webhook_id and invocation:
                session_id = client_wh.aiden.wait_for_webhook_session(
                    webhook_id, invocation
                )
                print(f"webhook session_id={session_id}")
                runs = client.aiden.list_executions(session_id=session_id, limit=1)
                if runs:
                    trace_id = str(runs[0].get("trace_id") or runs[0].get("id") or "")

        if not trace_id and schedule_id:
            runs = client.aiden.list_schedule_runs(schedule_id)
            for run in runs:
                tid = str(run.get("trace_id") or "")
                if tid:
                    trace_id = tid
                    break

        # Resolve only approvals that belong to this run.
        pending = client.aiden.list_approvals(status="pending")
        for item in pending:
            if not _approval_for_run(item, session_id, trace_id):
                continue
            approval_id = str(item.get("id") or "")
            if not approval_id:
                continue
            resolved = client.aiden.resolve_approval(
                approval_id, {"decision": "approve", "comment": "sdk demo auto-approve"}
            )
            print(f"resolved approval {approval_id}: {resolved.get('status')}")

        if not trace_id:
            print("no trace_id yet; check the Aiden UI for the run", file=sys.stderr)
            return 1

        subscribe_ok = False
        event_count = 0

        def _fallback_execution(reason: str) -> None:
            nonlocal subscribe_ok
            print(
                f"subscribe incomplete ({reason}); waiting on execution status",
                file=sys.stderr,
            )
            trace = client.aiden.wait_for_execution(trace_id)
            execution = trace.get("execution") if isinstance(trace, dict) else {}
            if not isinstance(execution, dict):
                execution = {}
            status = str(execution.get("status") or "")
            subscribe_ok = status in {"completed", "success", "succeeded"}
            print(f"subscribe complete ok={subscribe_ok} events={event_count}")
            print(f"execution status={status}")

        try:
            result = client.aiden.wait_for_subscribe_complete(
                trace_id, timeout_ms=args.timeout_seconds * 1000
            )
            subscribe_ok = bool(result.ok)
            event_count = int(result.event_count)
            print(f"subscribe complete ok={result.ok} events={result.event_count}")
            if not subscribe_ok:
                _fallback_execution(str(getattr(result, "raw_snippet", "") or "ok=False"))
        except Exception as err:  # noqa: BLE001
            message = str(err)
            if not any(
                token in message.lower()
                for token in ("timed out", "subscribeexecution", "aborted", "eof")
            ):
                raise
            _fallback_execution(message)
        tools = client.aiden.list_execution_tool_calls(trace_id, limit=5)
        print(f"recent tool calls: {len(tools)}")
        if not subscribe_ok:
            return 1
        return 0
    finally:
        # Cleanup demo resources
        if schedule_id:
            try:
                client.aiden.delete_schedule(schedule_id)
                print(f"deleted schedule {schedule_id}")
            except Exception as err:  # noqa: BLE001
                print(f"schedule cleanup: {err}", file=sys.stderr)
        if webhook_id:
            try:
                client.aiden.delete_webhook(webhook_id)
                print(f"deleted webhook {webhook_id}")
            except Exception as err:  # noqa: BLE001
                print(f"webhook cleanup: {err}", file=sys.stderr)


if __name__ == "__main__":
    raise SystemExit(main())