← All code samples

Ask → subscribe → session artifacts

You ask Aiden to do a job in plain language. When the run finishes, downloadable files show up on the session. This sample uses the public SDK step by step:

  1. startAsk / start_ask
  2. waitForSubscribeComplete / wait_for_subscribe_complete
  3. downloadSessionArtifacts / download_session_artifacts into a directory

Use this when you need all notes from a session, not one hard-coded filename. The helper lists metadata, sanitizes filenames, and downloads every named file. Prefer the single-file helpers only when you already know the name. The one-call helper Ask → artifact is enough when you only want session-report.md.

Ask journeys use the API token only (no webhook token).

When you need this

  • Operator follow-through. You ask Aiden to review recent org changes, leave a downloadable note, and optionally publish a Grafana dashboard when your integrations allow it. The note is a session artifact; a live dashboard may show up as a console_url in that note or in the run write-up.
  • Product changelog. You ask for an end-user note in Keep a Changelog form (Added, Changed, Deprecated, Removed, Fixed, Security). Download the markdown, keep it in CI artifacts, optionally ask for a Slack summary or a draft docs PR in the same message. Those side effects are part of the Ask; the SDK only starts, waits, and downloads.
  • Any multi-file session. Triage that ships both a markdown note and a JSON bundle. List first, then download what you care about.

Phrase Asks in ordinary language. Do not name internal tool ids in the message.

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.4.0rc11
python examples/ask/ask-session-artifacts.py \
  --message 'Look at recent org changes and leave me a markdown note I can download' \
  --out-dir ./out/session-artifacts

Optional pins: --agent my-ops-agent or --workflow-name incident-triage.

Minimal

from pathlib import Path
from stackgen import StackgenClient, StackgenConfig

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

start = client.aiden.start_ask(
    "Look at recent changes in our GitHub org. "
    "Leave me a short markdown note I can download.",
    entity_refs=[{"name": "my-ops-agent", "type": "agent"}],
)
session_id = start["session_id"]
trace_id = start["trace_id"]

done = client.aiden.wait_for_subscribe_complete(trace_id)
assert done.ok

downloaded = client.aiden.download_session_artifacts(
    session_id,
    "session-artifacts",
)
# Each entry is {"name", "path", "meta"}.
print([row["path"] for row in downloaded])

Keep a Changelog notes

Ask for sections from Keep a Changelog. Example message shape:

Write an end-user product changelog for last week's changes.
Use Keep a Changelog sections: Added, Changed, Deprecated, Removed, Fixed, and Security.
Skip empty sections. Credit authors when you can, and include PR links.
Leave that changelog as a markdown note I can download.
If Slack is available, post a short summary to #engineering.
If public docs would help, open a draft PR under docs/release-notes (do not merge).

After download, treat the file as ordinary markdown. The runnable scripts print which of those section headings appear. A minimal check:

import re

SECTIONS = ("Added", "Changed", "Deprecated", "Removed", "Fixed", "Security")

def keep_a_changelog_hits(text: str) -> list[str]:
    return [
        s for s in SECTIONS
        if re.search(r"^#{1,3}\s*" + re.escape(s) + r"\b", text, flags=re.M | re.I)
    ]

Empty weeks are fine. The note should say the window was quiet instead of inventing entries.

Operator follow-through and dashboards

Same SDK path. Change the Ask, not the client calls:

Look at what changed in our GitHub org since yesterday. You have Grafana too.
If those changes already have metrics, make a dashboard I can open, or update
one we already have if it fits. If something important shipped and there is
nothing to graph, and a small draft PR to add metrics would help, open one.
Leave me a short note I can download.

When a dashboard is published under your Grafana integration, published boards often land in an aiden-managed folder and the note (or run text) can include a console_url. The example scripts surface matching URLs from downloaded text. If no dashboard is appropriate, the downloadable note still completes the journey.

Complete scripts

#!/usr/bin/env python3
"""Ask → wait on subscribe → list and download session artifacts.

Use when you want every downloadable note from a finished Ask (operator
follow-through, product changelog, triage reports), not a single named file.

Docs: https://appcd-dev.github.io/stackgen-sdk/code-samples/ask-session-artifacts/

Usage:
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_WORKSPACE_ID=<workspace-uuid>
  python examples/ask/ask-session-artifacts.py \\
    --message 'Look at recent org changes and leave me a markdown note I can download'
  # optional pin:
  #   --agent my-ops-agent
  #   --out-dir ./out/session-artifacts
"""

from __future__ import annotations

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

from stackgen import StackgenClient, StackgenConfig, TimeoutError as SdkTimeoutError

# Keep a Changelog section titles (https://keepachangelog.com/).
KEEP_A_CHANGELOG_SECTIONS = (
    "Added",
    "Changed",
    "Deprecated",
    "Removed",
    "Fixed",
    "Security",
)


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 looks_like_text(name: str, data: bytes) -> bool:
    lower = name.lower()
    if re.search(r"\.(md|txt|markdown|json)$", lower):
        return True
    sample = data[:512].decode("utf-8", errors="replace")
    return bool(sample) and "\0" not in sample


def keep_a_changelog_hits(text: str) -> list[str]:
    hits: list[str] = []
    for section in KEEP_A_CHANGELOG_SECTIONS:
        if re.search(rf"^#{{1,3}}\s*{re.escape(section)}\b", text, flags=re.M | re.I):
            hits.append(section)
    return hits


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--message",
        default=(
            "Look at recent changes in our GitHub org. Leave me a short "
            "markdown note I can download covering what you found."
        ),
        help="Natural-language Ask message",
    )
    parser.add_argument("--agent", help="Pin an agent via entity_refs")
    parser.add_argument("--workflow-name", help="Pin a workflow via entity_refs")
    parser.add_argument(
        "--out-dir",
        default="session-artifacts",
        help="Directory for downloaded artifact files",
    )
    parser.add_argument(
        "--timeout-ms",
        type=int,
        default=30 * 60 * 1000,
        help="Subscribe wait budget in milliseconds",
    )
    args = parser.parse_args()

    client = StackgenClient(
        StackgenConfig(
            base_url=env("STACKGEN_URL"),
            api_token=env("STACKGEN_TOKEN"),
            workspace_id=env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
            timeout_seconds=max(30, args.timeout_ms // 1000),
        )
    )

    entity_refs: list[dict[str, str]] = []
    if args.agent:
        entity_refs.append({"name": args.agent, "type": "agent"})
    if args.workflow_name:
        entity_refs.append({"name": args.workflow_name, "type": "workflow"})

    session_id = ""
    try:
        start = client.aiden.start_ask(
            args.message,
            entity_refs=entity_refs or None,
        )
        session_id = str(start.get("session_id") or "")
        trace_id = str(start.get("trace_id") or "")
        if not session_id or not trace_id:
            raise SystemExit(f"start_ask missing session_id/trace_id: {start}")
        print(f"started session_id={session_id} trace_id={trace_id}", file=sys.stderr)

        try:
            done = client.aiden.wait_for_subscribe_complete(
                trace_id,
                timeout_ms=args.timeout_ms,
            )
            if not done.ok:
                print(
                    f"subscribe incomplete ({done}); waiting on execution status",
                    file=sys.stderr,
                )
                trace = client.aiden.wait_for_execution(trace_id)
                status = str((trace.get("execution") or {}).get("status") or "")
                print(f"execution status={status}", file=sys.stderr)
            else:
                print(f"subscribe complete ok={done.ok}", file=sys.stderr)
        except SdkTimeoutError as exc:
            print(
                f"subscribe timed out ({exc}); waiting on execution status",
                file=sys.stderr,
            )
            trace = client.aiden.wait_for_execution(trace_id)
            status = str((trace.get("execution") or {}).get("status") or "")
            print(f"execution status={status}", file=sys.stderr)

        out_dir = Path(args.out_dir)
        out_dir.mkdir(parents=True, exist_ok=True)

        downloaded = client.aiden.download_session_artifacts(session_id, str(out_dir))
        for file in downloaded:
            print(
                f"downloaded {file.get('name')} -> {file.get('path')}",
                file=sys.stderr,
            )

        if not downloaded:
            export = client.aiden.export_session(session_id, parts="report")
            parts = export.get("parts") or []
            report = next(
                (
                    str(part.get("content") or "")
                    for part in parts
                    if isinstance(part, dict) and part.get("kind") == "report"
                ),
                "",
            )
            if not report.strip():
                raise SystemExit(
                    f"no session artifacts and no export report for session_id={session_id}"
                )
            dest = out_dir / "session-report.md"
            dest.write_text(report, encoding="utf-8")
            downloaded.append(
                {"name": "session-report.md", "path": str(dest), "meta": {}}
            )
            print(f"wrote export report -> {dest}", file=sys.stderr)

        text_notes: list[str] = []
        for file in downloaded:
            data = Path(file["path"]).read_bytes()
            if not looks_like_text(file["name"], data):
                continue
            text = data.decode("utf-8", errors="replace").strip()
            if not text:
                continue
            text_notes.append(text)
            sections = keep_a_changelog_hits(text)
            if sections:
                print(
                    f"{file['name']}: Keep a Changelog sections found: "
                    + ", ".join(sections),
                    file=sys.stderr,
                )

        if text_notes:
            combined = out_dir / "combined-note.md"
            combined.write_text("\n\n---\n\n".join(text_notes), encoding="utf-8")
            print(f"wrote {combined}", file=sys.stderr)

        blob = "\n".join(text_notes)
        urls = re.findall(r"https?://[^\s)\]>]+", blob)
        console_urls = [
            u for u in urls if re.search(r"grafana|console_url|/d/", u, flags=re.I)
        ][:5]

        print(
            json.dumps(
                {
                    "session_id": session_id,
                    "trace_id": trace_id,
                    "artifacts": downloaded,
                    "console_urls": console_urls,
                },
                indent=2,
            )
        )
        return 0
    finally:
        if session_id:
            try:
                client.aiden.terminate_session(session_id, reason="sdk example finished")
            except Exception as exc:  # noqa: BLE001
                print(f"terminate session {session_id}: {exc}", file=sys.stderr)


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