← All code samples

Async webhook worker (complete)

Queue-friendly webhook automation: trigger with sg_aios_…, then poll and export with stackgen_…. Handles the common 202 without session_id case.

How to run · Minimal · Complete script

Step SDK call
1. Trigger trigger_webhook(payload_string)
2. Resolve session wait_for_webhook_session(webhook_id, invocation_id) when needed
3. Resume get_session
4. Write-up export_session(..., parts="report")

How to run

pip install stackgen-sdk

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

python examples/recipes/webhook-async-worker.py --start-only
python examples/recipes/webhook-async-worker.py

Minimal

import json
from stackgen import StackgenClient, StackgenConfig

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

trigger = client.aiden.trigger_webhook(
    json.dumps({"query": "Investigate the checkout CPU alert in prod", "correlation_id": "corr-123"})
)
session_id = trigger.get("session_id") or client.aiden.wait_for_webhook_session(
    trigger["webhook_id"],
    trigger["invocation_id"],
    dispatch_run_id=str(trigger.get("run_id") or ""),
)
export = client.aiden.export_session(session_id, parts="report")

Complete script

#!/usr/bin/env python3
"""Complete async webhook worker (obfuscated sample ids).

Trigger with the webhook token, then poll/run control with the API token.
Shows the dual-credential pattern queue workers need.

  pip install stackgen-sdk
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_WORKSPACE_ID=<workspace-uuid>
  export STACKGEN_WEBHOOK_TOKEN=sg_aios_…

  python webhook-async-worker.py --start-only
  python webhook-async-worker.py
"""

from __future__ import annotations

import argparse
import json
import os
import sys

from stackgen import StackgenClient, StackgenConfig
from stackgen.errors import StackgenError


def _env(*keys: str) -> str:
    for key in keys:
        value = os.environ.get(key, "").strip()
        if value:
            return value
    return ""


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--query",
        default="Investigate the checkout CPU alert in prod",
    )
    parser.add_argument("--correlation-id", default="corr-123")
    parser.add_argument("--timeout-seconds", type=int, default=1800)
    parser.add_argument("--poll-interval-seconds", type=int, default=5)
    parser.add_argument("--start-only", action="store_true")
    args = parser.parse_args()

    base = _env("STACKGEN_URL")
    token = _env("STACKGEN_TOKEN")
    org = _env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT", "STACKGEN_PROJECT_ID")
    webhook = _env("STACKGEN_WEBHOOK_TOKEN", "WEBHOOK_TOKEN")
    if not base or not token or not org or not webhook:
        raise SystemExit(
            "set STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID, "
            "and STACKGEN_WEBHOOK_TOKEN"
        )

    client = StackgenClient(
        StackgenConfig(
            base_url=base,
            api_token=token,
            workspace_id=org,
            webhook_token=webhook,
            timeout_seconds=args.timeout_seconds,
            poll_interval_seconds=args.poll_interval_seconds,
        )
    )

    # 1) Fire-and-forget trigger (webhook token). Body may be plain text or JSON.
    payload = json.dumps(
        {
            "query": args.query,
            "message": args.query,
            "correlation_id": args.correlation_id,
        }
    )
    triggered = client.aiden.trigger_webhook(payload)
    print("trigger:", json.dumps(triggered, indent=2, default=str))
    if args.start_only:
        return 0

    webhook_id = str(triggered.get("webhook_id") or "")
    invocation_id = str(triggered.get("invocation_id") or "")
    session_id = triggered.get("session_id")
    dispatch_run_id = str(triggered.get("run_id") or triggered.get("id") or "")

    # 2) If session is missing (HTTP 202), poll until session_id appears (API token)
    if not session_id:
        if not webhook_id or not invocation_id:
            print(
                "error: trigger missing webhook_id/invocation_id and session_id",
                file=sys.stderr,
            )
            return 1
        session_id = client.aiden.wait_for_webhook_session(
            webhook_id,
            invocation_id,
            dispatch_run_id=dispatch_run_id,
        )
        print(f"resolved session_id={session_id}", file=sys.stderr)

    session_id = str(session_id)
    detail = client.aiden.get_session(session_id)
    print("get_session:", json.dumps(detail, indent=2, default=str))

    export = client.aiden.export_session(session_id, parts="report")
    print("export:", json.dumps(export, indent=2, default=str))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except StackgenError as exc:
        print(f"error: {exc}", file=sys.stderr)
        raise SystemExit(1)

Related: Async webhook CI, Async Ask worker, Webhook → session → artifact.