specora 0.2.0rc1

Specora Platform SDK for Python - AI governance and policy enforcement

specora

Specora Platform SDK for Python - AI governance and policy enforcement.

Installation

pip install specora
# or
poetry add specora
# or
uv add specora

Quick Start

from specora import SpecoraClient

client = SpecoraClient(
    base_url="https://api.specora.ai",
    api_key="sk_...",
    vendor_id="vendor-uuid",
    tenant_id="tenant-uuid",
)

# Check policy before AI execution
policy = client.check_policy(
    subject={"type": "pr_run", "id": "run-123"},
)

if policy.decision == "block":
    print(f"Blocked by policy: {policy.severity}")
    exit(1)

# Execute your AI operation...

# Record the execution
result = client.record_execution({
    "provider": "openai",
    "model": "gpt-4",
    "cost_micros": 5000,  # $0.005
    "input_tokens": 1000,
    "output_tokens": 500,
    "success": True,
})

Async Client

from specora import AsyncSpecoraClient

async with AsyncSpecoraClient(
    base_url="https://api.specora.ai",
    api_key="sk_...",
    vendor_id="vendor-uuid",
    tenant_id="tenant-uuid",
) as client:
    policy = await client.check_policy(
        subject={"type": "pr_run", "id": "run-123"},
    )

Features

• Policy Evaluation: Check governance policies before AI operations
• Budget Management: Track and enforce spending limits
• Execution Recording: Record AI executions for tracking
• Assumptions Validation: Validate change assumptions
• Evidence Generation: Generate audit evidence bundles
• Risk Scoring: Get real-time risk metrics
• Webhook Verification: HMAC-SHA256 signature verification
• Async Support: Full async/await support with AsyncSpecoraClient

Purpose-Based Routing

Assign specific LLM providers/models to specific agents or task types. For example, route safety-rewriter to a local Ollama instance while routing style-assistant to Claude Sonnet.

auth = client.pre_authorize(
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    estimated_tokens=2000,
    estimated_cost_micros=500,
    routing_hints={
        "purpose": "safety-rewriter",
        "agent_name": "gtm-920-content-generator",
    },
)

# The server may override provider/model based on purpose routing policies
print(auth.selected_provider)  # e.g., "ollama"
print(auth.selected_model)     # e.g., "llama3:70b"

The org admin creates PURPOSE_ROUTING governance policies that map agents and task types to target providers/models. When pre_authorize() includes routing hints, the server applies matching purpose routing rules. If no fallback is configured and the target model is unavailable, the request is blocked (fail-closed).

Configuration

client = SpecoraClient(
    # Required
    base_url="https://api.specora.ai",
    api_key="sk_...",
    vendor_id="vendor-uuid",
    tenant_id="tenant-uuid",

    # Optional
    fail_open=False,  # Default: False (fail-closed)
    timeout=30.0,     # Request timeout in seconds
    retries=2,        # Retry attempts
)

Fail-Closed Semantics

By default, the SDK operates in fail-closed mode. This means:

• Network errors result in blocked decisions
• Server errors result in blocked decisions
• Timeouts result in blocked decisions

This ensures that if the Specora API is unavailable, your AI operations are blocked rather than allowed without governance.

To change this behavior:

client = SpecoraClient(
    # ...
    fail_open=True,  # Errors result in allowed decisions
)

API Reference

check_policy(subject, inputs=None)

Evaluate governance policy for an operation.

result = client.check_policy(
    subject={"type": "pr_run", "id": "run-123"},
    inputs={"context": "value"},  # optional
)

# result.decision: "allow" | "block" | "warn"
# result.severity: "low" | "medium" | "high" | "critical"
# result.invariants: list of InvariantResult
# result.override: OverrideEligibility

validate_budget(dimension="cost")

Check budget status for the tenant.

budget = client.validate_budget()

# budget.budget_remaining: float (USD)
# budget.budget_total: float (USD)
# budget.usage_percent: float (0-100)
# budget.forecast_breach: bool
# budget.enforcement_state: "normal" | "throttled" | "blocked"

record_execution(request)

Record an AI execution for budget tracking.

result = client.record_execution({
    "provider": "openai",
    "model": "gpt-4",
    "cost_micros": 5000,  # 1/1,000,000 USD
    "input_tokens": 1000,
    "output_tokens": 500,
    "duration_ms": 1500,
    "success": True,
    "external_run_id": "your-id",  # optional
    "metadata": {"key": "value"},  # optional
})

# result.execution_id: str
# result.budget_remaining_micros: int
# result.usage_percent: float
# result.enforcement_state: str

validate_assumptions(request)

Validate assumptions for a change.

from specora import Assumption

result = client.validate_assumptions({
    "run_id": "run-123",
    "assumptions": [
        {"id": "a1", "description": "No breaking changes", "impact_level": "high"},
    ],
})

# result.severity: "low" | "medium" | "high" | "critical"
# result.verification_required: bool
# result.merge_eligible: bool
# result.violations: list of AssumptionsViolation
# result.recommendations: list of str

generate_evidence_bundle(request)

Generate a signed evidence bundle.

evidence = client.generate_evidence_bundle({
    "run_id": "run-123",
    "artifact_types": ["confidence_report", "trace"],
    "include_provenance": True,
})

# evidence.bundle_id: str
# evidence.bundle_hash: str (sha256:...)
# evidence.storage_path: str

get_risk_score()

Get risk metrics for the tenant.

risk = client.get_risk_score()

# risk.risk_percentile: float (0-100)
# risk.drift_risk: float
# risk.override_frequency_risk: float
# risk.anomaly_status: "normal" | "elevated" | "critical"

verify_proof(manifest_hash) (PR-ENT-520-02)

Verify a proof by manifest hash.

result = client.verify_proof("abc123def456...")

# result.verified: bool
# result.proof_found: bool
# result.org_match: bool
# result.manifest_hash: str
# result.proof_type: str | None
# result.created_at: datetime | None
# result.published_root_match: bool | None
# result.manifest_spec_id: str | None  # e.g., "proof-manifest"
# result.manifest_schema_version: str | None  # e.g., "1.0.0"

Manifest Versioning (PR-ENT-520-02-PATCH-01): The response includes manifest_spec_id and manifest_schema_version for external verification. Store these values alongside the hash to enable offline verification with the correct contract version.

compute_proof_hash(data) (Offline Verification)

Compute a proof hash locally using canonical JSON serialization.

from specora import compute_proof_hash

# Compute hash matching server-side canonical JSON
hash_result = compute_proof_hash({
    "id": "...",
    "org_id": "...",
    "root_type": "ledger_daily",
    "root_hash": "...",
    "leaf_count": 100,
    "period_start": "2024-01-01T00:00:00+00:00",
    "period_end": "2024-01-31T23:59:59+00:00",
    "created_at": "2024-02-01T00:00:00+00:00",
})

print(f"Hash: {hash_result}")  # SHA-256 hex

Canonical JSON Rules:

• Sorted keys (alphabetically)
• Compact separators ("," and ":")
• ASCII-safe encoding (ensure_ascii=True)
• UTF-8 encoding for bytes

Webhook Verification

Verify incoming webhooks from Specora:

from flask import Flask, request, jsonify
from specora import verify_webhook_signature, parse_webhook_payload

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    is_valid = verify_webhook_signature(
        request.data,
        request.headers.get('X-Specora-Signature'),
        request.headers.get('X-Specora-Timestamp'),
        os.environ['WEBHOOK_SECRET'],
    )

    if not is_valid:
        return jsonify({'error': 'Invalid signature'}), 401

    payload = parse_webhook_payload(request.data)

    if payload.event_type == 'platform.budget_breach':
        # Handle budget breach
        pass
    elif payload.event_type == 'platform.invariant_failed':
        # Handle policy violation
        pass

    return jsonify({'received': True})

Webhook Event Types

• platform.invariant_failed - Governance invariant violation
• platform.budget_breach - Budget limit exceeded
• platform.provider_quarantined - AI provider quarantined
• platform.drift_detected - Model or schema drift
• platform.break_glass_used - Emergency override activated
• platform.sla_breach - SLA threshold violated
• platform.risk_score_changed - Significant risk change

Error Handling

The SDK throws typed exceptions for different failure scenarios:

from specora import (
    SpecoraClient,
    FailClosedError,
    BudgetExceededError,
    AuthenticationError,
    ValidationError,
)

try:
    result = client.check_policy(...)
except FailClosedError as e:
    # Network/server error in fail-closed mode
    print(f"Defaulting to blocked: {e.decision}")
except BudgetExceededError as e:
    # Budget exceeded
    print(f"Budget exceeded: {e.usage_percent}%")
except AuthenticationError:
    # Invalid API key
    pass
except ValidationError as e:
    # Invalid request parameters
    print(f"Validation errors: {e.validation_errors}")

Security

HTTPS Required

The SDK enforces HTTPS for all non-local base_url values. Plaintext HTTP would expose your API key (sent as a Bearer token on every request) to network observers.

# OK - production
SpecoraClient(base_url="https://api.specora.ai", ...)

# OK - local development
SpecoraClient(base_url="http://localhost:8765", ...)
SpecoraClient(base_url="http://127.0.0.1:8765", ...)

# RAISES ValueError - plaintext HTTP to remote host
SpecoraClient(base_url="http://api.specora.ai", ...)

Supply Chain

• Minimal runtime dependencies — only httpx and pydantic.
• No install-time scripts — pip install executes no lifecycle hooks.
• Fail-closed default — network/server errors block operations, not allow them.

Network Hardening

• No redirects — httpx is configured with follow_redirects=False to prevent the Authorization header from leaking to redirect targets.
• Response size cap — responses over 10 MB are rejected to prevent OOM from a malicious or compromised server.
• Retry-After cap — server-provided Retry-After values are capped at 120 seconds to prevent indefinite blocking.

Credential Handling

• API keys are only transmitted in the Authorization header over HTTPS.
• Keys are never logged or serialized to error messages.
• repr(client) returns SpecoraClient('https://...') — no API key exposed.
• Webhook verification uses constant-time comparison (hmac.compare_digest) to prevent timing attacks.
• Empty webhook secrets are rejected to prevent signature forgery.

Governance Router Safety

• Message size limits — max 1024 messages and 10 MB total content per request.
• Authorization expiry — checked before LLM execution to prevent stale authorizations.
• Conservative token estimation — uses chars/3 (not chars/4) to prevent budget overspend.
• Recording failures surfaced — recording_failed_warning field on response instead of silent error swallowing.

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT