OSSP Python SDK

CloudEvents-first client for emitting OSSP events with production-ready features like retry logic, validation, and trace context.

View on GitHub Get Started

Installation

Basic Installation

pip install ossp-sdk

Installs the core SDK with HTTP client and CloudEvents support.

With Validation

pip install "ossp-sdk[validate]"

Includes jsonschema for local payload validation against OSSP schemas.

Key Features

🔄

Retry Logic

Exponential backoff with jitter for reliable event delivery, handles retryable HTTP status codes (429, 5xx).

🔒

Idempotency

Built-in idempotency key support to prevent duplicate event processing.

🔍

Trace Context

W3C Trace Context propagation for distributed tracing integration.

✅

Schema Validation

Optional local validation against OSSP JSON schemas before sending.

⚡

Session Reuse

HTTP session pooling for improved performance in high-throughput scenarios.

📊

Observability

Structured logging and metrics for monitoring SDK performance.

Quick Start

Basic Usage

from ossp_sdk.client import OSSPClient

# Initialize client
client = OSSPClient(
    source_uri="urn:app:my-ai-system",
    collector_endpoint="https://collector.example.com",
    auth_token="your-auth-token"
)

# Emit a guardrail interaction event
event = client.emit(
    event_type="ai.safety.guardrail.interaction",
    resource={
        "model_id": "gpt-4o",
        "environment": "production",
        "model_version": "2025-09-01.3"
    },
    data={
        "action_taken": "block",
        "reason": "PII detected in user input",
        "severity": "high",
        "guardrail_id": "pii-filter-v2"
    },
    subject="urn:model:gpt-4o"
)

print(f"Event emitted: {event['id']}")

Configuration Options

Client Configuration

client = OSSPClient(
    collector_endpoint="https://api.example.com/events",
    auth_token="bearer-token",
    source_uri="urn:app:credit-scoring",
    timeout_seconds=5.0,
    max_retries=3,
    backoff_base_seconds=0.5,
    idempotency_key="unique-key-123"
)

Environment Variables

# Set via environment
export OSSP_COLLECTOR_ENDPOINT="https://api.example.com"
export OSSP_AUTH_TOKEN="your-token"
export OSSP_SOURCE_URI="urn:app:my-system"

# Enable validation
export OSSP_VALIDATE="true"
export OSSP_SCHEMA_DIR="/path/to/schemas"

Trace Context

# Propagate trace context
client.emit(
    event_type="ai.safety.guardrail.interaction",
    resource={"model_id": "gpt-4o", "environment": "prod"},
    data={"action_taken": "warn", "severity": "medium"},
    traceparent="00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
    tracestate="vendor=value"
)

Event Examples

Performance Drift

client.emit(
    event_type="ai.performance.drift.detected",
    resource={
        "model_id": "recommendation-engine-v2",
        "environment": "production"
    },
    data={
        "metric_name": "accuracy",
        "current_value": 0.82,
        "baseline_value": 0.91,
        "threshold": 0.05,
        "drift_type": "performance_degradation"
    }
)

Security Abuse Attempt

client.emit(
    event_type="ai.security.abuse.attempt",
    resource={
        "model_id": "chatbot-gpt4",
        "environment": "production"
    },
    data={
        "vector": "prompt_injection",
        "detector_id": "pi-detector-v1",
        "confidence": 0.92,
        "sample_id": "sample-789",
        "blocked": True
    }
)

Policy Violation

client.emit(
    event_type="ai.safety.policy.violation",
    resource={
        "model_id": "content-classifier",
        "environment": "production"
    },
    data={
        "policy_id": "content-safety-v3",
        "violation_type": "unsafe_content",
        "decision": "approved",
        "approver": "safety_team",
        "reason": "false positive - educational content"
    }
)

Advanced Features

Local Development

When no collector endpoint is configured, events are logged locally for development:

# No collector endpoint = local logging
client = OSSPClient(
    source_uri="urn:app:local-dev",
    collector_endpoint=None  # Events logged to stdout
)

# Events appear in logs
# [OSSP CE] {"specversion": "1.0", "id": "...", ...}

CLI Usage

Test event emission directly from the repository:

# From repository root
python sdk-python/examples/send_event.py --print

# With collector endpoint
python sdk-python/examples/send_event.py \
  --collector-endpoint https://api.example.com \
  --auth-token your-token \
  --log-level debug

Validation Setup

Enable local schema validation for development and testing:

# Install with validation
pip install "ossp-sdk[validate]"

# Point to schema directory
export OSSP_VALIDATE="true"
export OSSP_SCHEMA_DIR="./specification/schema/v1.0.0"

# Offline mode: vendor metaschemas
# specification/schema/meta/2020-12/

Production Considerations

Error Handling

  • Automatic retry with exponential backoff
  • Handles transient failures (429, 5xx status codes)
  • Graceful degradation on permanent failures
  • Comprehensive logging for troubleshooting

Performance

  • HTTP connection pooling via requests.Session
  • Configurable timeouts and retry limits
  • Minimal memory footprint
  • JSON serialization optimizations

Security

  • Bearer token authentication
  • HTTPS-only for production endpoints
  • No sensitive data logging
  • Configurable idempotency keys

Resources

GitHub Repository

Source code, issues, and contribution guidelines.

View Repository

Examples

Complete working examples and CLI tools.

Browse Examples

API Reference

Detailed API documentation and method signatures.

View Source