Stage0 authorization API docs

import os
import requests

STAGE0_URL = os.environ.get("STAGE0_BASE_URL", "<YOUR_STAGE0_BASE_URL>")
STAGE0_API_KEY = os.environ["STAGE0_API_KEY"]


def authorize_or_raise(goal: str, tool_name: str, side_effects: list[str], context: dict) -> dict:
    payload = {
        "goal": goal,
        "tools": [tool_name],
        "side_effects": side_effects,
        "constraints": ["require human approval for destructive actions"],
        "context": context,
    }
    response = requests.post(
        f"{STAGE0_URL}/check",
        headers={
            "x-api-key": STAGE0_API_KEY,
            "content-type": "application/json",
        },
        json=payload,
        timeout=5,
    )
    response.raise_for_status()
    decision = response.json()
    if decision["verdict"] != "ALLOW":
        raise RuntimeError(f"Stage0 blocked tool call: {decision}")
    return decision

# Call authorize_or_raise(...) immediately before each tool execution.

import requests
from typing import TypedDict
from langgraph.graph import StateGraph, END

STAGE0_URL = "<YOUR_STAGE0_BASE_URL>"
STAGE0_API_KEY = "<YOUR_API_KEY>"


class AgentState(TypedDict, total=False):
    goal: str
    tool_name: str
    side_effects: list[str]
    context: dict
    stage0: dict


def stage0_gate(state: AgentState) -> AgentState:
    response = requests.post(
        f"{STAGE0_URL}/check",
        headers={"x-api-key": STAGE0_API_KEY},
        json={
            "goal": state["goal"],
            "tools": [state["tool_name"]],
            "side_effects": state.get("side_effects", []),
            "context": state.get("context", {}),
        },
        timeout=5,
    )
    response.raise_for_status()
    return {"stage0": response.json()}


def route_after_gate(state: AgentState) -> str:
    verdict = state["stage0"]["verdict"]
    if verdict == "ALLOW":
        return "tools"
    if verdict == "DEFER":
        return "human_review"
    return "blocked"


graph = StateGraph(AgentState)
graph.add_node("stage0_gate", stage0_gate)
graph.add_conditional_edges("stage0_gate", route_after_gate)
graph.add_edge("blocked", END)

const STAGE0_URL = "<YOUR_STAGE0_BASE_URL>";
const STAGE0_API_KEY = process.env.STAGE0_API_KEY;

async function stage0Check(payload) {
  const response = await fetch(`${STAGE0_URL}/check`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "x-api-key": STAGE0_API_KEY,
    },
    body: JSON.stringify(payload),
  });

  if (!response.ok) {
    throw new Error(`Stage0 request failed: ${response.status}`);
  }

  return response.json();
}

server.tool("delete_customer", async (args) => {
  const decision = await stage0Check({
    goal: "Delete a customer record after support confirmation",
    tools: ["delete_customer"],
    side_effects: ["data deletion"],
    constraints: ["approval required", "business-hours only"],
    context: {
      actor_role: args.actorRole,
      approval_status: args.approvalStatus,
      environment: args.environment,
    },
  });

  if (decision.verdict !== "ALLOW") {
    return {
      content: [{ type: "text", text: JSON.stringify(decision, null, 2) }],
      isError: true,
    };
  }

  return performDelete(args);
});

# Pre-tool pattern (pseudo-code)
def before_tool_execution(tool_name, context):
    decision = stage0.check(goal, tools=[tool_name], side_effects=..., context=context)
    if decision.verdict != "ALLOW":
        raise AuthorizationError(decision)
    return decision  # Continue to tool execution

# Pre-webhook pattern
async def send_webhook(url, payload, context):
    decision = await stage0.check(
        goal=f"Send webhook to {url}",
        side_effects=["external notification"],
        constraints=["approval required for customer-facing channels"],
        context={**context, "approval_status": context.get("approval_status")}
    )
    if decision.verdict != "ALLOW":
        log.deferred("webhook", url, decision.request_id)
        return
    await http_post(url, payload)

# Pre-deploy pattern
def deploy_service(environment, service, context):
    decision = stage0.check(
        goal=f"Deploy {service} to {environment}",
        side_effects=["deploy"],
        constraints=[
            "required_role:platform_admin",
            f"allowed_environment:staging,production",
            "rollback_plan_required"
        ],
        context={**context, "environment": environment}
    )
    if decision.verdict != "ALLOW":
        return decision  # Do not proceed with deployment
    return kubectl_apply(service, environment)

# Acceptable fail-open pattern (read-only only)
async def safe_read_with_fallback(query, context):
    try:
        decision = await stage0.check(
            goal=query,
            tools=["db"],
            side_effects=[],  # No side effects - read only
            constraints=["read-only"],
            context=context,
            timeout_seconds=5
        )
        if decision.verdict == "ALLOW":
            return execute_query(query)
        # DENY or DEFER: respect the decision
        return None
    except (TimeoutError, ServiceUnavailableError):
        # Fail-open fallback for READ-ONLY operations only
        # This must be documented and monitored
        log.warn("stage0_unavailable_read_fallback", query)
        return execute_query(query)  # Proceed with read

# SAFE: Retry before side effect begins
async def check_with_retry(payload, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await stage0.check(payload, timeout_seconds=10)
        except NetworkError:
            if attempt == max_retries - 1:
                raise
            await sleep(2 ** attempt)  # Exponential backoff
    # If we reach here, check failed - side effect never started

# UNSAFE: Retry after side effect started (DO NOT DO THIS)
async def unsafe_pattern(goal, execute_fn):
    decision = await stage0.check(goal)
    if decision.verdict != "ALLOW":
        return
    result = execute_fn()  # Side effect starts here
    # If execution fails, DO NOT retry /check
    # The side effect may have partially completed
    # Handle execution errors separately, do not re-authorize

{
  "goal": "Deploy the payments service to production",
  "tools": ["argo_cd", "slack"],
  "side_effects": ["deployment", "external notification"],
  "constraints": [
    "approval required",
    "required_role:platform_admin",
    "allowed_environment:staging,production",
    "max_iterations:3",
    "timeout:300s",
    "max_cost_usd:5"
  ],
  "context": {
    "run_id": "run_9f3d7f51",
    "actor_role": "platform_admin",
    "approval_status": "approved",
    "approved_by": "[email protected]",
    "approved_at": "2026-03-14T09:30:00Z",
    "environment": "production",
    "current_iteration": 1,
    "elapsed_seconds": 42,
    "current_tool": "argo_cd",
    "recent_tools": ["github_actions"],
    "cumulative_cost_usd": 0.18
  },
  "pro": true
}