Quickstart

python3 docs/examples/python/playbooks_quickstart.py

#!/usr/bin/env python3
"""Minimal, dependency-free Playbooks quickstart.

Required environment:
  BLACKPEARL_API_BASE_URL  Configured gateway URL, for example http://localhost:8400
  BLACKPEARL_API_KEY       Project-scoped API key
"""
from __future__ import annotations

import json
import os
import random
import sys
import time
import uuid
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

TERMINAL = {"succeeded", "failed", "canceled"}
TRANSIENT_READ_CODES = {
    "rate_limit_exceeded",
    "internal_error",
    "audiences_backend_error",
    "audiences_backend_invalid_response",
    "audiences_backend_unavailable",
}


class APIError(RuntimeError):
    def __init__(self, status: int, code: str, message: str, retry_after: str | None = None):
        super().__init__(f"API error {status} {code}: {message}")
        self.status = status
        self.code = code
        self.retry_after = retry_after


def required_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"Set {name} before running this example")
    return value


def request_json(
    method: str,
    url: str,
    api_key: str,
    body: dict[str, Any] | None = None,
    extra_headers: dict[str, str] | None = None,
) -> tuple[dict[str, Any], Any]:
    payload = json.dumps(body).encode() if body is not None else None
    request = Request(
        url,
        data=payload,
        method=method,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json",
            **({"Content-Type": "application/json"} if body is not None else {}),
            **(extra_headers or {}),
        },
    )
    try:
        with urlopen(request, timeout=30) as response:
            return json.load(response), response.headers
    except HTTPError as exc:
        try:
            payload = json.loads(exc.read())
            error = payload.get("error", {})
        except (json.JSONDecodeError, AttributeError):
            error = {}
        raise APIError(
            exc.code,
            str(error.get("code", "http_error")),
            str(error.get("message", exc.reason)),
            exc.headers.get("Retry-After"),
        ) from exc
    except URLError as exc:
        raise RuntimeError(f"Could not reach the configured API: {exc.reason}") from exc


def persist_job_id(path: Path, job_id: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(f"{job_id}\n", encoding="utf-8")


def persistent_idempotency_key(path: Path) -> str:
    if path.exists() and path.read_text(encoding="utf-8").strip():
        return path.read_text(encoding="utf-8").strip()
    key = f"playbook-{uuid.uuid4()}"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(f"{key}\n", encoding="utf-8")
    return key


def create_or_resume(base_url: str, api_key: str, state_path: Path) -> str:
    if state_path.exists() and state_path.read_text(encoding="utf-8").strip():
        job_id = state_path.read_text(encoding="utf-8").strip()
        print(f"Resuming Playbooks job {job_id}")
        return job_id

    # Persist the key before sending. If the connection fails ambiguously, rerun
    # with this same request/key during the 24-hour replay window.
    idempotency_key = persistent_idempotency_key(
        state_path.with_name(f"{state_path.name}.idempotency")
    )
    job, _ = request_json(
        "POST",
        f"{base_url}/v1/playbooks",
        api_key,
        {"target_company_domain": "acme.example"},
        {"Idempotency-Key": idempotency_key},
    )
    job_id = job.get("id")
    if job.get("type") != "playbooks" or not isinstance(job_id, str) or not job_id:
        raise RuntimeError("Create response did not contain a typed Playbooks job ID")
    persist_job_id(state_path, job_id)
    print(f"Created Playbooks job {job_id}")
    return job_id


def poll_job(base_url: str, api_key: str, job_id: str) -> dict[str, Any]:
    timeout = float(os.environ.get("BLACKPEARL_POLL_DEADLINE_SECONDS", "120"))
    delay = float(os.environ.get("BLACKPEARL_POLL_INITIAL_SECONDS", "2"))
    max_delay = float(os.environ.get("BLACKPEARL_POLL_MAX_SECONDS", "10"))
    deadline = time.monotonic() + timeout

    while time.monotonic() < deadline:
        try:
            job, _ = request_json("GET", f"{base_url}/v1/jobs/{job_id}", api_key)
        except APIError as exc:
            if exc.code not in TRANSIENT_READ_CODES:
                raise
            if exc.code == "rate_limit_exceeded" and exc.retry_after:
                delay = max(delay, float(exc.retry_after))
            time.sleep(min(delay, max(0.0, deadline - time.monotonic())))
            delay = min(max_delay, max(delay * 2, 0.01))
            continue

        status = job.get("status")
        if status not in {"queued", "running", *TERMINAL}:
            raise RuntimeError(f"Unknown job status: {status!r}")
        print(f"status={status} progress={job.get('progress')}%")
        if status in TERMINAL:
            return job

        jittered = delay * random.uniform(0.8, 1.2)
        time.sleep(min(jittered, max(0.0, deadline - time.monotonic())))
        delay = min(max_delay, max(delay * 2, 0.01))

    raise TimeoutError(f"Polling deadline reached; resume job {job_id} from the saved state file")


def render_result(job: dict[str, Any]) -> None:
    status = job["status"]
    if status == "failed":
        raise RuntimeError(f"Playbooks job failed: {job.get('error') or 'No failure message returned'}")
    if status == "canceled":
        raise RuntimeError("Playbooks job was canceled")
    if job.get("type") != "playbooks" or status != "succeeded":
        raise RuntimeError("Expected a successful Playbooks job")

    result = job.get("result")
    if not isinstance(result, dict):
        raise RuntimeError("Successful Playbooks job did not contain a typed result")
    summary = result.get("business_summary")
    angles = result.get("sales_angles")
    if not isinstance(summary, str) or not summary or not isinstance(angles, list) or not angles or not isinstance(angles[0], str):
        raise RuntimeError("Playbooks result did not contain a useful business summary and sales angle")
    print(f"Business summary: {summary}")
    print(f"First sales angle: {angles[0]}")


def main() -> int:
    base_url = required_env("BLACKPEARL_API_BASE_URL").rstrip("/")
    api_key = required_env("BLACKPEARL_API_KEY")
    state_path = Path(os.environ.get("BLACKPEARL_JOB_STATE_FILE", ".blackpearl-playbooks-job"))
    try:
        job_id = create_or_resume(base_url, api_key, state_path)
        job = poll_job(base_url, api_key, job_id)
        render_result(job)
        state_path.unlink(missing_ok=True)
        state_path.with_name(f"{state_path.name}.idempotency").unlink(missing_ok=True)
        return 0
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        return 1


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