GaaS Python SDK

Python client library for the GaaS governance API. Provides async and sync clients with full type coverage.

Installation

pip install gaas-sdk

Quick Start

Async Client

from gaas_sdk import GaaSClient, build_intent, ActionType, TargetType

async with GaaSClient("https://api.gaas.is") as client:
    intent = build_intent(
        agent_id="my-agent",
        action_type=ActionType.COMMUNICATE,
        verb="send",
        target_type=TargetType.ENDPOINT,
        target_identifier="email-service",
        summary="Send welcome email to new user",
        content={"to": "user@example.com"},
    )
    response = await client.submit_intent(intent)
    print(response.data.verdict)  # APPROVE, BLOCK, ESCALATE, APPROVE_MODIFIED

Sync Client

from gaas_sdk import GaaSClientSync, build_intent, ActionType, TargetType

with GaaSClientSync("https://api.gaas.is") as client:
    intent = build_intent(
        agent_id="my-agent",
        action_type=ActionType.COMMUNICATE,
        verb="send",
        target_type=TargetType.ENDPOINT,
        target_identifier="email-service",
        summary="Send welcome email to new user",
        content={"to": "user@example.com"},
    )
    response = client.submit_intent(intent)
    print(response.data.verdict)

Authentication

Pass an API key via headers:

client = GaaSClient(
    "https://api.gaas.is",
    headers={"X-API-Key": "your-api-key"},
)

Client Configuration

Both GaaSClient (async) and GaaSClientSync accept the same configuration:

API Methods

Intent Submission

# Submit an intent for governance evaluation
response = await client.submit_intent(intent)

# Access the decision
response.data.verdict          # Verdict enum
response.data.reasoning        # DecisionReasoning
response.data.modifications    # List[Modification] (if APPROVE_MODIFIED)
response.data.risk_assessment  # RiskAssessment

Decision & Audit Retrieval

# Get a specific decision
decision = await client.get_decision("intent-id")

# Get the full audit record
audit = await client.get_audit("intent-id")

Health Check

health = await client.health_check()
print(health.data.status)   # "healthy"
print(health.data.version)  # "0.2.0"

Response Object

All methods return a GaaSResponse[T] with:

response.data   # The typed response payload (GovernanceDecision, AuditRecord, etc.)
response.meta   # ResponseMeta with request_id, decision_id, latency, status_code

Building Intents

The build_intent helper flattens the nested model tree into keyword arguments:

from gaas_sdk import build_intent, ActionType, TargetType, Sensitivity, Urgency

intent = build_intent(
    # Agent (required)
    agent_id="agent-001",
    agent_framework="custom",        # or AgentFramework.CUSTOM
    agent_name="My Agent",

    # Action (required)
    action_type=ActionType.EXECUTE,
    verb="deploy",

    # Target (required)
    target_type=TargetType.INFRASTRUCTURE,
    target_identifier="prod-cluster",
    target_sensitivity=Sensitivity.REGULATED,
    target_jurisdiction="US",

    # Payload (required)
    summary="Deploy model v2.1 to production cluster",
    content={"model": "v2.1", "cluster": "prod-us-east"},

    # Impact (optional)
    reversible=True,
    financial_exposure_usd=5000.0,
    audience_size=10000,
    regulatory_domains=["SOC2"],

    # Governance (optional)
    urgency=Urgency.HIGH,
    max_latency_ms=1000,
    require_deliberation=True,
)

For full control over nested models, construct IntentDeclaration directly:

from gaas_sdk import IntentDeclaration, AgentInfo, Action, ActionTarget, ActionPayload, EstimatedImpact

intent = IntentDeclaration(
    agent=AgentInfo(id="agent-001", framework="custom"),
    action=Action(
        type="EXECUTE",
        verb="deploy",
        target=ActionTarget(type="INFRASTRUCTURE", identifier="prod-cluster"),
    ),
    payload=ActionPayload(summary="Deploy model v2.1", content={"model": "v2.1"}),
    estimated_impact=EstimatedImpact(reversible=True, financial_exposure_usd=5000.0, audience_size=10000),
)

Field Filtering

Request only specific fields to reduce response size:

response = await client.submit_intent(
    intent,
    fields="verdict,reasoning.summary,risk_assessment.overall_score"
)

Webhook Integration

GaaS sends webhook events for decision lifecycle, escalations, learning observations, and system events. All webhooks are signed with HMAC-SHA256 for security.

Webhook Signature Validation

IMPORTANT: Always validate webhook signatures before processing events to prevent spoofing attacks.

import hmac
import hashlib
import json
from typing import Any

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify HMAC-SHA256 webhook signature using constant-time comparison.

    Args:
        payload: Raw request body bytes (not parsed JSON)
        signature: Value from X-GaaS-Signature header (format: "sha256=<hex>")
        secret: Webhook secret from registration (starts with "whsec_")

    Returns:
        True if signature is valid, False otherwise
    """
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(f"sha256={expected}", signature)

Webhook Event Handling

from flask import Flask, request, jsonify

app = Flask(__name__)

# Store your webhook secret securely (from webhook registration response)
WEBHOOK_SECRET = "whsec_abc123..."

@app.route("/webhooks/gaas", methods=["POST"])
def handle_gaas_webhook():
    # Get raw request body before parsing
    payload_bytes = request.get_data()
    signature = request.headers.get("X-GaaS-Signature")
    event_type = request.headers.get("X-GaaS-Event")
    delivery_id = request.headers.get("X-GaaS-Delivery")

    # Validate signature before processing
    if not verify_webhook_signature(payload_bytes, signature, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401

    # Parse event payload
    event = json.loads(payload_bytes)

    # Route to appropriate handler based on event type
    handlers = {
        "decision.approved": handle_decision_approved,
        "decision.blocked": handle_decision_blocked,
        "decision.escalated": handle_decision_escalated,
        "escalation.decided": handle_escalation_decided,
        "escalation.timed_out": handle_escalation_timed_out,
        "escalation.cancelled": handle_escalation_cancelled,
        "escalation.reassigned": handle_escalation_reassigned,
        "observation.recorded": handle_observation_recorded,
        "quota.exceeded": handle_quota_exceeded,
        "rate_limit.exceeded": handle_rate_limit_exceeded,
    }

    handler = handlers.get(event_type)
    if handler:
        handler(event)
    else:
        print(f"Unknown event type: {event_type}")

    # Return 200 to acknowledge receipt
    return jsonify({"status": "received"}), 200


def handle_decision_approved(event: dict[str, Any]) -> None:
    """Handle decision.approved events."""
    decision_id = event["decision_id"]
    intent_id = event["intent_id"]
    data = event["data"]

    print(f"Decision {decision_id} approved for intent {intent_id}")
    # Update your database, notify agent, etc.


def handle_decision_blocked(event: dict[str, Any]) -> None:
    """Handle decision.blocked events."""
    decision_id = event["decision_id"]
    intent_id = event["intent_id"]
    data = event["data"]

    print(f"Decision {decision_id} blocked for intent {intent_id}")
    # Alert agent owner, log for audit, etc.


def handle_escalation_decided(event: dict[str, Any]) -> None:
    """Handle escalation.decided events."""
    escalation_id = event["escalation_id"]
    escalation_status = event["escalation_status"]
    data = event["data"]

    print(f"Escalation {escalation_id} decided with status: {escalation_status}")
    # Notify agent, update workflow, etc.


def handle_quota_exceeded(event: dict[str, Any]) -> None:
    """Handle quota.exceeded events."""
    org_id = event["organization_id"]
    data = event["data"]

    print(f"Quota exceeded for organization {org_id}")
    # Send upgrade notification, throttle agent, etc.

Event Types

GaaS sends webhooks for these event types:

Decision Lifecycle

• decision.approved - Intent was approved
• decision.blocked - Intent was blocked
• decision.escalated - Intent was escalated for human review

Escalation Lifecycle

• escalation.decided - Escalation was reviewed and decided
• escalation.timed_out - Escalation timed out without decision
• escalation.cancelled - Escalation was cancelled
• escalation.reassigned - Escalation was reassigned to another reviewer

Learning & Patterns

• observation.recorded - New learning observation recorded
• policy.calibrated - Policy weights/thresholds were calibrated
• pattern.detected - Behavioral pattern detected

System Events

• quota.exceeded - Monthly quota limit exceeded
• rate_limit.exceeded - Rate limit threshold exceeded

Webhook Payload Structure

All webhook payloads follow this structure:

{
    "event_type": "decision.approved",
    "timestamp": "2026-02-14T12:00:00Z",
    "webhook_id": "wh_abc123",
    "organization_id": "org_default",
    "data": {
        # Event-specific data (full decision, escalation, etc.)
    },
    # Event-specific fields (optional):
    "decision_id": "dec_xyz789",
    "intent_id": "int_123abc",
    "escalation_id": "esc_def456",
    "observation_id": "obs_ghi789",
}

Security Best Practices

1. Always validate signatures - Reject webhooks with invalid signatures
2. Use constant-time comparison - Prevents timing attacks (use hmac.compare_digest)
3. Store secrets securely - Use environment variables or secret managers
4. Validate event structure - Check required fields before processing
5. Idempotency - Use X-GaaS-Delivery header to deduplicate events
6. Return 200 quickly - Process events asynchronously if they're slow
7. Handle retries gracefully - GaaS retries failed deliveries (3 attempts max)

FastAPI Example

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/webhooks/gaas")
async def handle_webhook(request: Request):
    payload_bytes = await request.body()
    signature = request.headers.get("x-gaas-signature")

    if not verify_webhook_signature(payload_bytes, signature, WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()
    # Process event...

    return {"status": "received"}

Error Handling

from gaas_sdk import (
    GaaSError,                 # Base exception
    GaaSValidationError,       # 422 — invalid request structure
    GaaSSemanticError,         # 422 — semantic validation failures
    GaaSAuthenticationError,   # 401 — invalid API key
    GaaSQuotaExceededError,    # 402 — monthly quota exceeded
    GaaSNotFoundError,         # 404 — resource not found
    GaaSConflictError,         # 409 — idempotency conflict
    GaaSRateLimitError,        # 429 — rate limit exceeded
    GaaSServerError,           # 500 — server error
    GaaSConnectionError,       # Network/connection failure
)

try:
    response = await client.submit_intent(intent)
except GaaSAuthenticationError:
    print("Invalid API key")
except GaaSQuotaExceededError as e:
    print(f"Quota exceeded. Upgrade plan or wait until: {e.details.reset_time}")
except GaaSRateLimitError as e:
    # Exponential backoff recommended
    retry_after = e.details.get("retry_after_seconds", 60)
    await asyncio.sleep(retry_after)
except GaaSValidationError as e:
    print(f"Validation failed: {e.message}")
    print(f"Details: {e.details}")
except GaaSConnectionError:
    print("Could not reach GaaS server")

Rate Limit Handling Best Practices

import asyncio
from gaas_sdk import GaaSRateLimitError

async def submit_with_retry(client, intent, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await client.submit_intent(intent)
        except GaaSRateLimitError as e:
            if attempt == max_retries - 1:
                raise
            retry_after = e.details.get("retry_after_seconds", 60)
            wait_time = min(retry_after * (2 ** attempt), 300)  # Cap at 5 minutes
            await asyncio.sleep(wait_time)

Exported Models & Enums

The SDK re-exports all models needed to build intents and interpret decisions:

Models: IntentDeclaration, GovernanceDecision, AuditRecord, AgentInfo, Action, ActionTarget, ActionPayload, EstimatedImpact, GovernanceRequest, ContextProvided, RiskAssessment, Modification, BlockDetail, EscalationDetail, DecisionReasoning

Enums: ActionType, TargetType, Sensitivity, AgentFramework, TrustTier, Urgency, FallbackOnTimeout, DataCategory, Verdict, ModificationType, RiskClassification

Requirements

• Python 3.11+
• httpx >= 0.27.0
• pydantic >= 2.9.0

Development

pip install -e ".[dev]"
pytest

License

Apache-2.0