#!/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= 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())