Metadata-Version: 2.4
Name: gaas-sdk
Version: 0.2.7
Summary: Python SDK for GaaS (Governance as a Service)
Project-URL: Homepage, https://gaas.is
Project-URL: Documentation, https://gaas.to/docs/sdks.html
Project-URL: Repository, https://github.com/H2OmAI/gaas
Project-URL: Changelog, https://github.com/H2OmAI/gaas/blob/main/CHANGELOG.md
Author-email: H2Om <sdk@gaas.is>
License-Expression: Apache-2.0
Keywords: agents,ai,gaas,governance,sdk
Classifier: Development Status :: 4 - Beta
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.9.0
Provides-Extra: dev
Requires-Dist: fastapi>=0.115.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24.0; extra == 'dev'
Requires-Dist: pytest>=8.3.0; extra == 'dev'
Description-Content-Type: text/markdown

# GaaS Python SDK

Python client library for the [GaaS](https://github.com/H2OmAI/gaas) governance API. Provides async and sync clients with full type coverage.

## Installation

```bash
pip install gaas-sdk
```

## Quick Start

### Async Client

```python
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.API_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

```python
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.API_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:

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

## Client Configuration

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

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `base_url` | `str` | `"https://api.gaas.is"` | GaaS API server URL |
| `api_version` | `str` | `"v1"` | API version prefix |
| `timeout` | `float` | `30.0` | Request timeout in seconds |
| `headers` | `dict` | `None` | Additional HTTP headers |
| `http_client` | `httpx.AsyncClient` | `None` | Custom httpx client (async only) |

## API Methods

### Intent Submission

```python
# 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

```python
# 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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.

```python
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

```python
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:

```python
{
    "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

```python
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

```python
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

```python
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

```bash
pip install -e ".[dev]"
pytest
```

## License

Apache-2.0
