← All code samples

Webhook → session → artifact

The most common CI pattern: trigger a webhook, wait for the workflow session, then download a session artifact (usually session-report.md).

Webhook automation needs two tokens: a webhook token (sg_aios_…) to POST the trigger, and an API token (stackgen_…) to wait for artifacts and download reports.

flowchart LR
  A[Alert or JSON body] --> B[POST webhooks/trigger]
  B --> C{session_id on 202?}
  C -->|yes| D[Poll session artifacts]
  C -->|no| E[Poll getWebhookRun]
  E --> D
  D --> F[Download session-report.md]
Step What the SDK does
1. Trigger POST /guild/api/v1/webhooks/trigger with webhook token
2. Session Use session_id from 202 when present; otherwise poll run detail until set
3. Artifact Poll session artifacts, then download to a local path

Complete scripts

Minimal — one SDK call (recommended)

from stackgen import StackgenClient, StackgenConfig

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

result = client.aiden.run_webhook_and_download_report(
    "P1 | checkout-api error rate above SLO"
)
print(result.session_id, result.output_path)

Runnable CLI scripts

Full command-line examples with flags for CI pipelines.

#!/usr/bin/env python3
"""Webhook trigger → wait for session → download session artifact (CI-friendly).

Documented on https://appcd-dev.github.io/stackgen-sdk/code-samples/webhook-session-artifact/
under **Webhook → session → artifact**.

Usage:

  export WEBHOOK_TOKEN='sg_aios_…'
  python webhook-download-report.py \\
    --base-url https://app.stackgen.com \\
    --api-token stackgen_… \\
    --workspace-id <workspace-uuid> \\
    --query 'P1 | checkout-api error rate above SLO'
"""

from __future__ import annotations

import argparse
import os
import sys

from stackgen import StackgenClient, StackgenConfig


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Trigger a webhook, wait for session_id, download session-report.md.",
    )
    parser.add_argument("--base-url", default=os.environ.get("STACKGEN_URL", ""), help="Mothership URL (no /guild suffix)")
    parser.add_argument("--api-token", default=os.environ.get("STACKGEN_TOKEN", ""), help="API token (stackgen_…)")
    parser.add_argument(
        "--workspace-id",
        "--org-id",
        default=os.environ.get("STACKGEN_WORKSPACE_ID") or os.environ.get("STACKGEN_PROJECT") or os.environ.get("STACKGEN_PROJECT_ID", ""),
        help="Workspace UUID (sent as orgId)",
    )
    parser.add_argument(
        "--webhook-token",
        default="",
        help="Webhook token (sg_aios_…); or set WEBHOOK_TOKEN",
    )
    parser.add_argument("--query", required=True, help="Alert text or payload for the webhook")
    parser.add_argument(
        "--artifact-name",
        default="session-report.md",
        help="Session artifact filename to download",
    )
    parser.add_argument(
        "--output-path",
        default="session-report.md",
        help="Local path to write the artifact",
    )
    parser.add_argument("--timeout-seconds", type=int, default=1800)
    parser.add_argument("--poll-interval-seconds", type=int, default=5)
    args = parser.parse_args()

    webhook_token = (
        args.webhook_token
        or os.environ.get("STACKGEN_WEBHOOK_TOKEN", "")
        or os.environ.get("WEBHOOK_TOKEN", "")
    )
    if not args.base_url or not args.api_token or not args.workspace_id:
        print(
            "error: set --base-url/--api-token/--workspace-id or STACKGEN_URL/STACKGEN_TOKEN/STACKGEN_WORKSPACE_ID",
            file=sys.stderr,
        )
        return 2
    if not webhook_token:
        print("error: set --webhook-token or WEBHOOK_TOKEN", file=sys.stderr)
        return 2

    client = StackgenClient(
        StackgenConfig(
            base_url=args.base_url,
            api_token=args.api_token,
            webhook_token=webhook_token,
            workspace_id=args.workspace_id,
            artifact_name=args.artifact_name,
            output_path=args.output_path,
            timeout_seconds=args.timeout_seconds,
            poll_interval_seconds=args.poll_interval_seconds,
        )
    )
    result = client.aiden.run_webhook_and_download_report(
        args.query,
        artifact_name=args.artifact_name,
        output_path=args.output_path,
    )
    print(
        f"webhook_id={result.webhook_id} session_id={result.session_id} "
        f"output={result.output_path}",
        file=sys.stderr,
    )
    return 0


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

Step-by-step (advanced)

Use this when you need the raw trigger response, custom artifact waiting, or to download the artifact in a separate step. Pass --download-report to add step 3 on the same run.

#!/usr/bin/env python3
"""Webhook trigger → resolve session_id (optional artifact download).

Documented at https://appcd-dev.github.io/stackgen-sdk/code-samples/webhook-session-artifact/

Usage:

  export WEBHOOK_TOKEN='sg_aios_…'
  python webhook-wait-session.py \\
    --base-url https://app.stackgen.com \\
    --api-token stackgen_… \\
    --workspace-id <workspace-uuid> \\
    --json-body \\
    --query 'P1 | checkout-api error rate above SLO'

  # Also download session-report.md when the workflow finishes:
  python webhook-wait-session.py … --download-report
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from pathlib import Path

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


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default=os.environ.get("STACKGEN_URL", ""))
    parser.add_argument("--api-token", default=os.environ.get("STACKGEN_TOKEN", ""))
    parser.add_argument(
        "--workspace-id",
        "--org-id",
        default=os.environ.get("STACKGEN_WORKSPACE_ID") or os.environ.get("STACKGEN_PROJECT")
        or os.environ.get("STACKGEN_PROJECT_ID", ""),
    )
    parser.add_argument("--webhook-token", default="", help="Or set WEBHOOK_TOKEN env")
    parser.add_argument("--query", default="Test webhook from stackgen-sdk example")
    parser.add_argument("--json-body", action="store_true")
    parser.add_argument("--download-report", action="store_true")
    parser.add_argument("--artifact-name", default="session-report.md")
    parser.add_argument("--output-path", default="session-report.md")
    parser.add_argument("--timeout-seconds", type=int, default=1800)
    parser.add_argument("--poll-interval-seconds", type=int, default=5)
    args = parser.parse_args()

    webhook_token = (
        args.webhook_token
        or os.environ.get("STACKGEN_WEBHOOK_TOKEN", "")
        or os.environ.get("WEBHOOK_TOKEN", "")
    )
    if not args.base_url or not args.api_token or not args.workspace_id:
        print(
            "error: set --base-url/--api-token/--workspace-id or "
            "STACKGEN_URL/STACKGEN_TOKEN/STACKGEN_WORKSPACE_ID",
            file=sys.stderr,
        )
        return 2
    if not webhook_token:
        print("error: set --webhook-token or WEBHOOK_TOKEN", file=sys.stderr)
        return 2

    client = StackgenClient(
        StackgenConfig(
            base_url=args.base_url,
            api_token=args.api_token,
            webhook_token=webhook_token,
            workspace_id=args.workspace_id,
            artifact_name=args.artifact_name,
            output_path=args.output_path,
            timeout_seconds=args.timeout_seconds,
            poll_interval_seconds=args.poll_interval_seconds,
        )
    )

    payload = (
        json.dumps({"message": args.query, "query": args.query})
        if args.json_body
        else args.query
    )
    trigger = client.aiden.trigger_webhook(payload)
    print(json.dumps(trigger, indent=2, default=str))

    session_id = str(trigger.get("session_id") or "")
    webhook_id = str(trigger.get("webhook_id") or "")
    invocation_id = str(trigger.get("invocation_id") or "")
    dispatch_run_id = str(trigger.get("run_id") or trigger.get("id") or "")
    if not session_id:
        if not webhook_id or not invocation_id:
            raise StackgenError("trigger response missing session_id and run ids")
        session_id = client.aiden.wait_for_webhook_session(
            webhook_id,
            invocation_id,
            dispatch_run_id=dispatch_run_id,
        )

    print(f"session_id={session_id}", file=sys.stderr)

    if args.download_report:
        want = args.artifact_name.casefold()
        deadline = time.monotonic() + args.timeout_seconds
        while time.monotonic() < deadline:
            listed = client.aiden.list_session_artifacts(session_id)
            if any(
                str(item.get("name") or "").casefold() == want for item in listed
            ):
                break
            time.sleep(args.poll_interval_seconds)
        else:
            raise TimeoutError(args.artifact_name, args.timeout_seconds)
        path = client.aiden.download_session_artifact(
            session_id, args.artifact_name, args.output_path
        )
        if not Path(path).is_file():
            raise StackgenError(f"download missing file {path}")
        print(f"output_path={path}", file=sys.stderr)

    return 0


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

Related: Webhook → session only — same trigger flow without downloading.