SDK & Code Examples

Python SDK Wrapper

A lightweight Python client that handles authentication, retries, and pagination:

# StackFlow Python SDK wrapper
import time
import random
import boto3
import requests
from typing import Optional, Dict, Any, List

class StackFlowClient:
    def __init__(self, instance_url: str, email: str, password: str,
                 client_id: str = "570cnagpgoochn29a113du6jnt",
                 region: str = "us-east-1"):
        self.base_url = f"{instance_url}/prod/api"
        self._token = None
        self._token_expiry = 0
        self._email = email
        self._password = password
        self._client_id = client_id
        self._cognito = boto3.client("cognito-idp", region_name=region)
        self.session = requests.Session()

    def _refresh_token(self):
        resp = self._cognito.initiate_auth(
            AuthFlow="USER_PASSWORD_AUTH",
            AuthParameters={"USERNAME": self._email, "PASSWORD": self._password},
            ClientId=self._client_id
        )
        self._token = resp["AuthenticationResult"]["IdToken"]
        self._token_expiry = time.time() + 3500  # 1hr - 100s buffer
        self.session.headers.update({"Authorization": f"Bearer {self._token}"})

    def _ensure_token(self):
        if not self._token or time.time() >= self._token_expiry:
            self._refresh_token()

    def _request(self, method: str, path: str, **kwargs) -> requests.Response:
        self._ensure_token()
        url = f"{self.base_url}{path}"
        for attempt in range(5):
            resp = self.session.request(method, url, **kwargs)
            if resp.status_code == 429:
                wait = int(resp.headers.get("Retry-After", 60))
                time.sleep(wait)
                continue
            if resp.status_code >= 500:
                time.sleep(2**attempt + random.random())
                continue
            return resp
        return resp

    # ── Incidents ──
    def list_incidents(self, **params) -> Dict:
        return self._request("GET", "/incidents", params=params).json()

    def get_incident(self, incident_id: str) -> Dict:
        return self._request("GET", f"/incidents/{incident_id}").json()

    def create_incident(self, data: Dict) -> Dict:
        return self._request("POST", "/incidents", json=data).json()

    def update_incident(self, incident_id: str, data: Dict) -> Dict:
        return self._request("PATCH", f"/incidents/{incident_id}", json=data).json()

    def add_work_note(self, incident_id: str, body: str, visibility: str = "team") -> Dict:
        return self._request("POST", f"/incidents/{incident_id}/notes",
                            json={"body": body, "visibility": visibility}).json()

    # ── Changes ──
    def list_changes(self, **params) -> Dict:
        return self._request("GET", "/changes", params=params).json()

    def create_change(self, data: Dict) -> Dict:
        return self._request("POST", "/changes", json=data).json()

    # ── Knowledge ──
    def ask(self, question: str, model: str = "claude-3-sonnet") -> Dict:
        return self._request("POST", "/ai/ask",
                           json={"question": question, "model": model}).json()

    # ── Pagination helper ──
    def paginate(self, path: str, **params) -> List[Dict]:
        results = []
        params.setdefault("limit", 100)
        params["page"] = 1
        while True:
            resp = self._request("GET", path, params=params).json()
            results.extend(resp.get("data", []))
            if not resp.get("has_more"):
                break
            params["page"] += 1
        return results


# Usage:
# client = StackFlowClient("https://your-instance.stackflow-tech.com", "you@example.com", "pass")
# open_p1s = client.list_incidents(priority="P1", state="new")
# all_incidents = client.paginate("/incidents", state="in_progress")

Node.js Helper Class

const {
  CognitoIdentityProviderClient,
  InitiateAuthCommand
} = require("@aws-sdk/client-cognito-identity-provider");

class StackFlowClient {
  constructor({ instanceUrl, email, password, clientId = "570cnagpgoochn29a113du6jnt" }) {
    this.baseUrl = `${instanceUrl}/prod/api`;
    this.email = email;
    this.password = password;
    this.clientId = clientId;
    this.token = null;
    this.tokenExpiry = 0;
    this.cognito = new CognitoIdentityProviderClient({ region: "us-east-1" });
  }

  async ensureToken() {
    if (this.token && Date.now() < this.tokenExpiry) return;
    const cmd = new InitiateAuthCommand({
      AuthFlow: "USER_PASSWORD_AUTH",
      ClientId: this.clientId,
      AuthParameters: { USERNAME: this.email, PASSWORD: this.password }
    });
    const resp = await this.cognito.send(cmd);
    this.token = resp.AuthenticationResult.IdToken;
    this.tokenExpiry = Date.now() + 3500000; // 1hr - 100s
  }

  async request(method, path, body, params) {
    await this.ensureToken();
    const url = new URL(`${this.baseUrl}${path}`);
    if (params) Object.entries(params).forEach(([k,v]) => url.searchParams.set(k, v));
    for (let attempt = 0; attempt < 5; attempt++) {
      const resp = await fetch(url.toString(), {
        method,
        headers: { "Authorization": `Bearer ${this.token}`, "Content-Type": "application/json" },
        body: body ? JSON.stringify(body) : undefined
      });
      if (resp.status === 429) {
        const wait = parseInt(resp.headers.get("Retry-After") || "60") * 1000;
        await new Promise(r => setTimeout(r, wait));
        continue;
      }
      if (resp.status >= 500 && attempt < 4) {
        await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 1000));
        continue;
      }
      return resp.json();
    }
  }

  // Incidents
  listIncidents(params) { return this.request("GET", "/incidents", null, params); }
  getIncident(id) { return this.request("GET", `/incidents/${id}`); }
  createIncident(data) { return this.request("POST", "/incidents", data); }
  updateIncident(id, data) { return this.request("PATCH", `/incidents/${id}`, data); }

  // AI
  ask(question, model = "claude-3-sonnet") {
    return this.request("POST", "/ai/ask", { question, model });
  }
}

// Usage:
// const sf = new StackFlowClient({ instanceUrl: "https://your-instance.stackflow-tech.com", email: "you@example.com", password: "pass" });
// const incidents = await sf.listIncidents({ priority: "P1", state: "new" });

Postman Collection

Import the StackFlow API Postman collection for interactive API exploration:

{
  "info": {
    "name": "StackFlow API",
    "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
  },
  "variable": [
    {"key": "base_url", "value": "https://your-instance.stackflow-tech.com/prod/api"},
    {"key": "id_token", "value": ""},
    {"key": "cognito_url", "value": "https://stackflow-identity-373544523367.auth.us-east-1.amazoncognito.com"}
  ],
  "auth": {
    "type": "bearer",
    "bearer": [{"key": "token", "value": "{{id_token}}"}]
  },
  "item": [
    {
      "name": "Get Token",
      "request": {
        "method": "POST",
        "url": "{{cognito_url}}/oauth2/token",
        "body": {"mode": "urlencoded", "urlencoded": [
          {"key": "grant_type", "value": "password"},
          {"key": "client_id", "value": "570cnagpgoochn29a113du6jnt"},
          {"key": "username", "value": "{{email}}"},
          {"key": "password", "value": "{{password}}"}
        ]}
      }
    },
    {
      "name": "List Incidents",
      "request": {"method": "GET", "url": "{{base_url}}/incidents"}
    }
  ]
}

Save this as stackflow-postman.json and import via Postman → Import → Raw text.

Common Patterns

# Pattern: Auto-create incident from monitoring alert
def handle_cloudwatch_alarm(alarm_data: dict, sf_client: StackFlowClient):
    priority_map = {"CRITICAL": "P1", "HIGH": "P2", "MEDIUM": "P3", "LOW": "P4"}
    incident = sf_client.create_incident({
        "short_description": f"CloudWatch Alarm: {alarm_data['AlarmName']}",
        "description": alarm_data.get("AlarmDescription", ""),
        "priority": priority_map.get(alarm_data.get("severity", "MEDIUM"), "P3"),
        "category": "monitoring",
        "subcategory": "cloudwatch",
        "assignment_group": "Platform Engineering",
        "ci_id": alarm_data.get("stackflow_ci_id")
    })
    return incident["id"]

# Pattern: Bulk close resolved incidents older than 30 days
import datetime
cutoff = (datetime.datetime.utcnow() - datetime.timedelta(days=30)).isoformat() + "Z"
old_resolved = sf_client.paginate("/incidents", state="resolved", created_before=cutoff)
ids = [i["id"] for i in old_resolved]
sf_client._request("POST", "/incidents/bulk", json={"operation": "update", "ids": ids, "fields": {"state": "closed"}})

Environment Variables

Variable	Value	Description
STACKFLOW_BASE_URL	https://your-instance.stackflow-tech.com/prod/api	API base URL
STACKFLOW_CLIENT_ID	570cnagpgoochn29a113du6jnt	Cognito app client ID
STACKFLOW_COGNITO_URL	https://stackflow-identity-373544523367.auth.us-east-1.amazoncognito.com	Cognito hosted UI URL
STACKFLOW_USER_POOL_ID	us-east-1_WKK1AVJ2m	Cognito user pool ID
STACKFLOW_API_KEY	sfk_prod_...	Static API key (alternative to JWT)