← All code samples

Async Ask worker (complete)

Queue-friendly Ask automation: start on a worker, persist session_id / trace_id, then follow up from a scheduler with a one-shot get_run_status / getRunStatus. Do not block the submit path on wait_for_execution.

Sample labels are fictional Acme Corp ids (acme-incident-triage, ACME-1001, acme-corr-123).

How to run · Minimal · Complete script

Step SDK call Blocks?
1. Start start_ask / startAsk with entity_refs workflow pin no
2. Follow-up get_run_status / getRunStatus no (one shot)
3. Download download_session_artifact when ready no

How to run

pip install stackgen-sdk

export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_PROJECT=<project-uuid>
# optional: export STACKGEN_WORKFLOW=acme-incident-triage

python examples/recipes/ask-async-worker.py --start-only
python examples/recipes/ask-async-worker.py --follow-up \
  --session-id <uuid> --trace-id <id>

Minimal

from stackgen import StackgenClient, StackgenConfig
from stackgen.aiden import RunPhase

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

# Worker — fire and forget
start = client.aiden.start_ask(
    "Investigate the Acme checkout CPU alert in prod",
    entity_refs=[{"name": "acme-incident-triage", "type": "workflow"}],
    source_metadata={"correlation_id": "acme-corr-123", "incident_id": "ACME-1001"},
)
# Persist start["session_id"] and start["trace_id"] in your task store.

# Scheduler — one shot (repeat until ready)
status = client.aiden.get_run_status(
    session_id=start["session_id"],
    trace_id=start["trace_id"],
    ready_artifact="session-report.md",
)
if status.phase is RunPhase.ERROR:
    raise SystemExit("run failed")
if status.ready:
    client.aiden.download_session_artifact(
        status.session_id, "session-report.md", "session-report.md"
    )

Complete script

Full CLI from examples/recipes/ (embedded at docs build time).

#!/usr/bin/env python3
"""Async Ask worker — fire-and-forget start + separate follow-up.

Submit on one process; poll from another with get_run_status (never blocks for
the full investigation). Sample labels are fictional Acme Corp ids.

  pip install stackgen-sdk
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_PROJECT=<project-uuid>

  # Worker: start and exit immediately (persist the printed ids)
  python ask-async-worker.py --start-only

  # Scheduler tick: one-shot status (exit 2 = still running)
  python ask-async-worker.py --follow-up \\
    --session-id <uuid> --trace-id <id>
"""

from __future__ import annotations

import argparse
import json
import os
import sys

from stackgen import StackgenClient, StackgenConfig
from stackgen.aiden import RunPhase
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 _client() -> StackgenClient:
    base = _env("STACKGEN_URL")
    token = _env("STACKGEN_TOKEN")
    org = _env("STACKGEN_PROJECT", "STACKGEN_PROJECT_ID")
    if not base or not token or not org:
        raise SystemExit(
            "set STACKGEN_URL, STACKGEN_TOKEN, and STACKGEN_PROJECT"
        )
    return StackgenClient(
        StackgenConfig(
            base_url=base,
            api_token=token,
            org_id=org,
        )
    )


def _start(args: argparse.Namespace) -> int:
    client = _client()
    start = client.aiden.start_ask(
        args.message,
        entity_refs=[{"name": args.workflow, "type": "workflow"}],
        source_app="acme-automation",
        source_metadata={
            "correlation_id": args.correlation_id,
            "incident_id": args.incident_id,
        },
    )
    payload = {
        "session_id": start.get("session_id"),
        "trace_id": start.get("trace_id"),
        "run_id": start.get("run_id"),
        "workflow": args.workflow,
        "correlation_id": args.correlation_id,
        "incident_id": args.incident_id,
    }
    print(json.dumps(payload, indent=2, default=str))
    if not payload["session_id"] and not payload["trace_id"]:
        print("error: start missing session_id and trace_id", file=sys.stderr)
        return 1
    return 0


def _follow_up(args: argparse.Namespace) -> int:
    if not args.session_id and not args.trace_id:
        print("error: --follow-up requires --session-id and/or --trace-id", file=sys.stderr)
        return 1
    client = _client()
    status = client.aiden.get_run_status(
        session_id=args.session_id or None,
        trace_id=args.trace_id or None,
        ready_artifact=args.ready_artifact or None,
    )
    print(json.dumps(status.to_dict(), indent=2))
    if status.phase is RunPhase.ERROR:
        return 1

    ready_ok = True
    if args.ready_artifact:
        ready_ok = status.ready
    elif status.phase in {RunPhase.PENDING, RunPhase.RUNNING, RunPhase.UNKNOWN}:
        ready_ok = False

    if (
        ready_ok
        and args.download_dir
        and status.session_id
        and (not args.ready_artifact or status.ready)
    ):
        from pathlib import Path

        out = Path(args.download_dir)
        out.mkdir(parents=True, exist_ok=True)
        name = args.ready_artifact or "session-report.md"
        if name in status.artifact_names:
            dest = out / name
            client.aiden.download_session_artifact(status.session_id, name, str(dest))
            print(f"downloaded {dest}", file=sys.stderr)

    return 0 if ready_ok else 2


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--workflow",
        default=_env("STACKGEN_WORKFLOW") or "acme-incident-triage",
    )
    parser.add_argument(
        "--message",
        default="Investigate the Acme checkout CPU alert in prod",
    )
    parser.add_argument("--correlation-id", default="acme-corr-123")
    parser.add_argument("--incident-id", default="ACME-1001")
    parser.add_argument(
        "--start-only",
        action="store_true",
        help="Fire-and-forget start; print ids and exit",
    )
    parser.add_argument(
        "--follow-up",
        action="store_true",
        help="One-shot get_run_status for a previously started run",
    )
    parser.add_argument("--session-id", default="")
    parser.add_argument("--trace-id", default="")
    parser.add_argument(
        "--ready-artifact",
        default="session-report.md",
        help="Artifact that means the run is consumable",
    )
    parser.add_argument(
        "--download-dir",
        default="",
        help="When following up and ready, download artifact here",
    )
    args = parser.parse_args()

    if args.follow_up:
        return _follow_up(args)
    # Default and --start-only both start without waiting.
    return _start(args)


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

Related: Automation surface, Ask → export report, Async webhook worker, FAQ — queue / scheduler.