SRE → investigate alert
Start an SRE investigation from an alert that already exists in the inbox, then wait until the investigation has RCA fields. Requires SRE on the workspace and an API token.
| Step | What the SDK does |
|---|---|
| 1. List or get alert | list_alerts / get_alert |
| 2. Investigate and wait | investigate_alert_and_wait (starts, then polls) |
| 3. Optional | Use guild_session_id with Aiden export APIs for the session file |
Prefer this path when the alert is already in SRE and you want the investigation record plus structured RCA. Use Ask → artifact when you only have a free-text prompt and no alert id. To refresh the inbox first, see Pull alerts.
Investigation status
| Status | Meaning |
|---|---|
queued, running |
Still working. Keep waiting. |
completed |
RCA is ready. Read structured_rca and result_summary. |
draft |
Triage saved with gaps. Read triage_metadata (more_info_requested, more_info_notes). |
resolved |
Operator closed the thread. |
failed, cancelled |
Stop. The wait helper raises. |
Minimal — list, investigate, wait
from stackgen import StackgenClient, StackgenConfig
client = StackgenClient(
StackgenConfig(
base_url="https://app.stackgen.com",
api_token="stackgen_…",
workspace_id="<workspace-uuid>",
)
)
alerts = client.sre.list_alerts(status="active", q="checkout CPU")
rows = alerts.get("alerts") or alerts.get("items") or []
alert_id = rows[0]["id"]
inv = client.sre.investigate_alert_and_wait(
alert_id,
operator_message="Investigate checkout CPU in prod",
)
print(inv.get("result_summary"))
rca = inv.get("structured_rca") or {}
print((rca.get("plain_summary") or {}).get("verdict"))
print(inv.get("guild_session_id"))import { StackgenClient, StackgenConfig } from "stackgen-sdk";
const client = new StackgenClient(
new StackgenConfig({
baseUrl: "https://app.stackgen.com",
apiToken: "stackgen_…",
workspaceId: "<workspace-uuid>",
}),
);
const alerts = (await client.sre.listAlerts({
status: "active",
q: "checkout CPU",
})) as { alerts?: Array<{ id: string }>; items?: Array<{ id: string }> };
const alertId = alerts.alerts?.[0]?.id ?? alerts.items?.[0]?.id;
if (!alertId) {
throw new Error("no matching alert");
}
const inv = await client.sre.investigateAlertAndWait(alertId, {}, {
operatorMessage: "Investigate checkout CPU in prod",
});
console.log(inv.result_summary);
const rca = (inv.structured_rca ?? {}) as Record<string, unknown>;
console.log((rca.plain_summary as Record<string, unknown> | undefined)?.verdict);
console.log(inv.guild_session_id);alerts, err := client.SRE.ListAlerts(ctx, map[string]string{
"status": "active",
"q": "checkout CPU",
})
if err != nil { return err }
body := alerts.(map[string]any)
items, _ := body["alerts"].([]any)
if len(items) == 0 {
items, _ = body["items"].([]any)
}
if len(items) == 0 {
return fmt.Errorf("no matching alert")
}
alertID := fmt.Sprint(items[0].(map[string]any)["id"])
inv, err := client.SRE.InvestigateAlertAndWait(ctx, alertID, map[string]any{
"operator_message": "Investigate checkout CPU in prod",
})
if err != nil { return err }
fmt.Println(inv["result_summary"])
fmt.Println(inv["guild_session_id"])
return nilHow to run
export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
python examples/sre/investigate-alert.py --q 'checkout CPU' --pollexport STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
npx tsx examples/sre/investigate-alert.ts --q 'checkout CPU' --pollexport STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
go run ./examples/go/sre/investigate-alert.go --q 'checkout CPU' --pollAdd --list-only to check access without starting an investigation.
Complete script
- Python:
examples/sre/investigate-alert.py - TypeScript:
examples/sre/investigate-alert.ts - Go:
examples/go/sre/investigate-alert.go
#!/usr/bin/env python3
"""SRE: list alerts → investigate → wait for the RCA.
Proves the alert-already-in-inbox path. Use --list-only when the tenant
has no alerts yet (HTTP 200 + empty list is success).
Documented at:
https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-investigate-alert/
Credentials (flags override env):
STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID (or STACKGEN_ORG_ID / STACKGEN_PROJECT)
Usage:
pip install -e ./clients/python
python examples/sre/investigate-alert.py --list-only
python examples/sre/investigate-alert.py --q 'checkout' --poll
python examples/sre/investigate-alert.py --alert-id <uuid> --poll
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from stackgen import StackgenClient, StackgenConfig
from stackgen.errors import StackgenError
def _org_id(cli_value: str | None) -> str:
return (
(cli_value or "").strip()
or os.environ.get("STACKGEN_WORKSPACE_ID", "").strip()
or os.environ.get("STACKGEN_ORG_ID", "").strip()
or os.environ.get("STACKGEN_PROJECT", "").strip()
or os.environ.get("STACKGEN_PROJECT_ID", "").strip()
)
def _config(args: argparse.Namespace) -> StackgenConfig:
base = (args.base_url or os.environ.get("STACKGEN_URL", "")).strip()
token = (args.api_token or os.environ.get("STACKGEN_TOKEN", "")).strip()
org = _org_id(args.workspace_id)
if not base or not token:
raise SystemExit(
"set --base-url/--api-token or STACKGEN_URL / STACKGEN_TOKEN"
)
return StackgenConfig(
base_url=base,
api_token=token,
workspace_id=org,
timeout_seconds=args.timeout_seconds,
poll_interval_seconds=args.poll_interval_seconds,
)
def _alert_rows(payload: object) -> list[dict]:
if not isinstance(payload, dict):
return []
rows = payload.get("alerts") or payload.get("items") or []
return [r for r in rows if isinstance(r, dict)]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-url", default="")
parser.add_argument("--api-token", default="")
parser.add_argument("--workspace-id",
"--org-id", default="")
parser.add_argument("--status", default="active", help="Alert status filter")
parser.add_argument("--q", default="checkout", help="Alert search query")
parser.add_argument("--alert-id", default="", help="Skip list; investigate this id")
parser.add_argument(
"--list-only",
action="store_true",
help="List matching alerts and exit 0 (empty list is OK)",
)
parser.add_argument(
"--poll",
action="store_true",
help="After investigate, wait until the RCA is ready",
)
parser.add_argument(
"--export-session",
action="store_true",
help="If guild_session_id is set, call aiden.export_session(parts=report)",
)
parser.add_argument("--timeout-seconds", type=int, default=1800)
parser.add_argument("--poll-interval-seconds", type=int, default=5)
parser.add_argument(
"--operator-message",
default="Investigate checkout CPU in prod",
)
args = parser.parse_args()
client = StackgenClient(_config(args))
alert_id = args.alert_id.strip()
if not alert_id:
alerts = client.sre.list_alerts(status=args.status, q=args.q)
if not isinstance(alerts, dict):
print(f"error: unexpected list_alerts shape: {alerts!r}", file=sys.stderr)
return 1
items = _alert_rows(alerts)
print("list_alerts:", json.dumps(alerts, indent=2, default=str))
print(f"count={len(items)}", file=sys.stderr)
if args.list_only:
return 0
if not items:
print(
"error: no matching alert (pass --alert-id or use --list-only)",
file=sys.stderr,
)
return 1
alert_id = str(items[0].get("id") or "")
if not alert_id:
print("error: alert item missing id", file=sys.stderr)
return 1
if not args.poll:
started = client.sre.investigate_alert(
alert_id,
operator_message=args.operator_message,
)
print("investigate:", json.dumps(started, indent=2, default=str))
return 0
investigation = client.sre.investigate_alert_and_wait(
alert_id,
operator_message=args.operator_message,
)
print("investigation:", json.dumps(investigation, indent=2, default=str))
plain = investigation.get("structured_rca") if isinstance(investigation, dict) else None
if isinstance(plain, dict):
summary = plain.get("plain_summary")
if isinstance(summary, dict):
print(
"plain_summary:",
json.dumps(summary, indent=2, default=str),
)
root = plain.get("root_cause")
if root:
print(f"root_cause={root}", file=sys.stderr)
session_id = investigation.get("guild_session_id") if isinstance(investigation, dict) else None
if args.export_session and session_id:
export = client.aiden.export_session(str(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)
if "401" in str(exc) or "missing identity" in str(exc).lower():
print(
"hint: SRE /app/sre routes need the edge to inject workspace "
"identity (X-Appcd-Org). If Aiden works but SRE returns 401, "
"the mothership app proxy is not forwarding PAT identity.",
file=sys.stderr,
)
raise SystemExit(1)/**
* SRE: list alerts → investigate → wait for the RCA.
*
* Proves the alert-already-in-inbox path. Use --list-only when the tenant
* has no alerts yet (HTTP 200 + empty list is success).
*
* Documented at:
* https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-investigate-alert/
*
* Credentials (flags override env):
* STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID (or STACKGEN_ORG_ID / STACKGEN_PROJECT)
*
* Usage:
* npx tsx examples/sre/investigate-alert.ts --list-only
* npx tsx examples/sre/investigate-alert.ts --q 'checkout' --poll
* npx tsx examples/sre/investigate-alert.ts --alert-id <uuid> --poll
*/
import { parseArgs } from "node:util";
import { StackgenClient, StackgenConfig } from "stackgen-sdk";
function asRecord(value: unknown): Record<string, unknown> | undefined {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: undefined;
}
function show(label: string, value: unknown): void {
console.log(`${label}:`, JSON.stringify(value, null, 2));
}
function alertRows(payload: unknown): Record<string, unknown>[] {
const rec = asRecord(payload);
const rows = (rec?.alerts ?? rec?.items ?? []) as unknown[];
return rows.filter(
(r): r is Record<string, unknown> => !!r && typeof r === "object" && !Array.isArray(r),
);
}
async function main(): Promise<number> {
const { values } = parseArgs({
options: {
"base-url": { type: "string", default: "" },
"api-token": { type: "string", default: "" },
"org-id": { type: "string" },
"workspace-id": { type: "string", default: "" },
status: { type: "string", default: "active" },
q: { type: "string", default: "checkout" },
"alert-id": { type: "string", default: "" },
"list-only": { type: "boolean", default: false },
poll: { type: "boolean", default: false },
"export-session": { type: "boolean", default: false },
"timeout-seconds": { type: "string", default: "1800" },
"poll-interval-seconds": { type: "string", default: "5" },
"operator-message": { type: "string", default: "Investigate checkout CPU in prod" },
help: { type: "boolean", short: "h", default: false },
},
});
if (values.help) {
console.log(
"usage: investigate-alert.ts [--list-only] [--q TEXT] [--status S] " +
"[--alert-id ID] [--poll] [--export-session] [--operator-message TEXT]",
);
return 0;
}
const baseUrl = (values["base-url"] || process.env.STACKGEN_URL || "").trim();
const apiToken = (values["api-token"] || process.env.STACKGEN_TOKEN || "").trim();
const orgId = (
(values["org-id"] || values["workspace-id"]) ||
process.env.STACKGEN_WORKSPACE_ID ||
process.env.STACKGEN_ORG_ID ||
process.env.STACKGEN_PROJECT ||
process.env.STACKGEN_PROJECT_ID ||
""
).trim();
if (!baseUrl || !apiToken) {
console.error("set --base-url/--api-token or STACKGEN_URL / STACKGEN_TOKEN");
return 2;
}
const client = new StackgenClient(
new StackgenConfig({
baseUrl,
apiToken,
workspaceId: orgId,
timeoutSeconds: Number(values["timeout-seconds"]),
pollIntervalSeconds: Number(values["poll-interval-seconds"]),
}),
);
let alertId = (values["alert-id"] ?? "").trim();
if (!alertId) {
const alerts = await client.sre.listAlerts({ status: values.status, q: values.q });
const items = alertRows(alerts);
show("list_alerts", alerts);
console.error(`count=${items.length}`);
if (values["list-only"]) {
return 0;
}
if (!items[0]?.id) {
console.error("error: no matching alert (pass --alert-id or use --list-only)");
return 1;
}
alertId = String(items[0].id);
}
if (!values.poll) {
show(
"investigate",
await client.sre.investigateAlert(alertId, {}, {
operatorMessage: values["operator-message"],
}),
);
return 0;
}
const investigation = await client.sre.investigateAlertAndWait(
alertId,
{},
{ operatorMessage: values["operator-message"] },
);
show("investigation", investigation);
const rca = asRecord(investigation.structured_rca);
const plain = asRecord(rca?.plain_summary);
if (plain) {
show("plain_summary", plain);
}
if (rca?.root_cause) {
console.error(`root_cause=${rca.root_cause}`);
}
const sessionId = investigation.guild_session_id;
if (values["export-session"] && sessionId) {
show("export", await client.aiden.exportSession(String(sessionId), { parts: "report" }));
}
return 0;
}
main()
.then((code) => process.exit(code))
.catch((err: unknown) => {
const text = String(err instanceof Error ? err.message : err);
console.error(`error: ${text}`);
if (text.includes("401") || text.toLowerCase().includes("missing identity")) {
console.error(
"hint: SRE /app/sre routes need the edge to inject workspace identity " +
"(X-Appcd-Org). If Aiden works but SRE returns 401, the mothership " +
"app proxy is not forwarding PAT identity.",
);
}
process.exit(1);
});// List SRE alerts and optionally start an investigation, waiting for the RCA.
//
// Documented at https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-investigate-alert/
//
// go run ./examples/go/sre/investigate-alert.go --list-only --q checkout
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"strings"
stackgen "github.com/appcd-dev/stackgen-sdk/clients/go"
)
func envOr(keys ...string) string {
for _, key := range keys {
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v
}
}
return ""
}
func main() {
baseURL := flag.String("base-url", "", "Mothership URL")
apiToken := flag.String("api-token", "", "API token")
orgID := flag.String("workspace-id", "", "Workspace UUID (sent as orgId)")
flag.StringVar(orgID, "org-id", *orgID, "Deprecated: use -workspace-id")
status := flag.String("status", "active", "Alert status filter")
query := flag.String("q", "checkout", "Alert search query")
alertID := flag.String("alert-id", "", "Skip list; investigate this id")
listOnly := flag.Bool("list-only", false, "List matching alerts and exit")
poll := flag.Bool("poll", false, "Wait until the RCA is ready")
exportSession := flag.Bool("export-session", false, "Export the Aiden session report when present")
timeoutSeconds := flag.Int("timeout-seconds", 1800, "Wait timeout")
pollInterval := flag.Int("poll-interval-seconds", 5, "Poll interval")
operatorMessage := flag.String("operator-message", "Investigate checkout CPU in prod", "Message sent with investigate")
flag.Parse()
base := strings.TrimSpace(*baseURL)
if base == "" {
base = envOr("STACKGEN_URL")
}
token := strings.TrimSpace(*apiToken)
if token == "" {
token = envOr("STACKGEN_TOKEN")
}
org := strings.TrimSpace(*orgID)
if org == "" {
org = envOr("STACKGEN_WORKSPACE_ID", "STACKGEN_ORG_ID", "STACKGEN_PROJECT", "STACKGEN_PROJECT_ID")
}
if base == "" || token == "" {
fmt.Fprintln(os.Stderr, "set STACKGEN_URL and STACKGEN_TOKEN")
os.Exit(2)
}
client, err := stackgen.NewClient(stackgen.Config{
BaseURL: base,
APIToken: token,
WorkspaceID: org,
TimeoutSeconds: *timeoutSeconds,
PollIntervalSeconds: *pollInterval,
})
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
ctx := context.Background()
id := strings.TrimSpace(*alertID)
if id == "" {
alerts, err := client.SRE.ListAlerts(ctx, map[string]string{"status": *status, "q": *query})
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ := json.MarshalIndent(alerts, "", " ")
fmt.Println("list_alerts:", string(raw))
if *listOnly {
return
}
body, _ := alerts.(map[string]any)
items, _ := body["alerts"].([]any)
if len(items) == 0 {
items, _ = body["items"].([]any)
}
if len(items) == 0 {
fmt.Fprintln(os.Stderr, "error: no matching alert (pass --alert-id or use --list-only)")
os.Exit(1)
}
first, _ := items[0].(map[string]any)
id, _ = first["id"].(string)
if id == "" {
fmt.Fprintln(os.Stderr, "error: alert item missing id")
os.Exit(1)
}
}
body := map[string]any{"operator_message": *operatorMessage}
if !*poll {
started, err := client.SRE.InvestigateAlert(ctx, id, body)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ := json.MarshalIndent(started, "", " ")
fmt.Println("investigate:", string(raw))
return
}
investigation, err := client.SRE.InvestigateAlertAndWait(ctx, id, body)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ := json.MarshalIndent(investigation, "", " ")
fmt.Println("investigation:", string(raw))
if rca, ok := investigation["structured_rca"].(map[string]any); ok {
if plain, ok := rca["plain_summary"].(map[string]any); ok {
p, _ := json.MarshalIndent(plain, "", " ")
fmt.Println("plain_summary:", string(p))
}
if root := fmt.Sprint(rca["root_cause"]); root != "" && root != "<nil>" {
fmt.Fprintln(os.Stderr, "root_cause=", root)
}
}
if *exportSession {
if sessionID, _ := investigation["guild_session_id"].(string); sessionID != "" {
export, err := client.Aiden.ExportSession(ctx, sessionID, "report", "")
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ = json.MarshalIndent(export, "", " ")
fmt.Println("export:", string(raw))
}
}
}Notes
investigate_alert_and_waitstarts the run and polls untilcompleted,draft, orresolved. It raises onfailedorcancelled.- If the alert id is unknown, the API returns 404. Pull alerts first with Pull alerts.
- Optional fields:
force_new,operator_message,remediation_requested,mode. - When
guild_session_idis set, downloadsession-report.mdor callclient.aiden.export_session(session_id, parts="report").
Related: Pull alerts · Triage alerts · Ask → artifact