doopal 1.1.2

Python SDK for Doopal AI Governance Platform

Doopal AI Governance Python SDK

The Doopal Python SDK provides a simple and powerful way to integrate AI governance, redaction, and policy enforcement into your Python applications. Route your LLM API calls through Doopal to ensure compliance, security, and cost control.

Installation

pip install doopal

Quick Start

Async Usage (Recommended)

import asyncio
from doopal_client import DoopalClient

async def main():
    async with DoopalClient(api_key="your-api-key") as client:
        response = await client.chat_completion(
            provider="openai",
            model="gpt-3.5-turbo",
            messages=[
                {"role": "user", "content": "Hello, how are you?"}
            ]
        )
        print(response.response)

asyncio.run(main())

Synchronous Usage

from doopal_client import DoopalSyncClient

client = DoopalSyncClient(api_key="your-api-key")

response = client.chat_completion(
    provider="openai",
    model="gpt-3.5-turbo",
    messages=[
        {"role": "user", "content": "Hello, how are you?"}
    ]
)

print(response.response)
client.close()

Features

Core AI Gateway

• Multi-Provider Support: Works with OpenAI, Anthropic, Azure OpenAI, Cohere, and more
• Governance & Compliance: Automatic policy enforcement and compliance checking
• Content Redaction: Sensitive data detection and redaction
• Cost Control: Budget management and usage tracking
• Streaming Support: Real-time streaming responses
• Error Handling: Comprehensive error handling with specific exception types
• Async/Sync: Both asynchronous and synchronous interfaces

Enterprise Features (v1.1.0+)

• Organization Management: Multi-tenant organization administration
• Role-Based Access Control (RBAC): User roles and permissions management
• Single Sign-On (SSO): SAML/OIDC enterprise authentication
• Advanced Analytics: Compliance, threat detection, performance, and cost analytics
• Audit Logging: Comprehensive audit trails and compliance reporting
• Security Monitoring: Real-time threat detection and alerting
• Policy Management: Advanced policy creation, testing, and management
• Provider Management: AI provider configuration and monitoring
• Comprehensive Health Checks: Kubernetes-ready health, readiness, and liveness probes
• Data Validation: Structured data and PII validation capabilities

Configuration

Environment Variables

export DOOPAL_API_KEY="your-api-key"
export DOOPAL_BASE_URL="https://api.doopal.com"  # Optional
export DOOPAL_ORG_ID="your-org-id"  # Optional for multi-tenant setups

Client Configuration

client = DoopalClient(
    api_key="your-api-key",
    base_url="https://api.doopal.com",  # Default
    timeout=30,  # Request timeout in seconds
    max_retries=3,  # Maximum retry attempts
    organization_id="your-org-id"  # Optional
)

API Reference

Chat Completions

response = await client.chat_completion(
    provider="openai",
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of France?"}
    ],
    temperature=0.7,
    max_tokens=100,
    stream=False
)

Streaming Chat Completions

async for chunk in await client.chat_completion(
    provider="openai",
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True
):
    print(chunk, end="")

Text Completions

response = await client.completion(
    provider="openai",
    model="gpt-3.5-turbo-instruct",
    prompt="The capital of France is",
    max_tokens=50
)

Embeddings

response = await client.embeddings(
    provider="openai",
    model="text-embedding-ada-002",
    input_text="Hello world"
)

Usage Statistics

stats = await client.get_usage_stats()
print(f"Total requests: {stats['total_requests']}")
print(f"Total cost: ${stats['total_cost']}")

Health Check

# Basic health check
health = await client.health_check()
print(f"Status: {health['status']}")

# Detailed health status (includes database, redis, etc.)
detailed_health = await client.get_detailed_health()
print(f"Database: {detailed_health['database']['status']}")
print(f"Redis: {detailed_health['redis']['status']}")

# Kubernetes readiness probe
readiness = await client.get_readiness_status()
print(f"Ready: {readiness['status']}")

# Kubernetes liveness probe
liveness = await client.get_liveness_status()
print(f"Live: {liveness['status']}")

Enterprise Organization Management

# Get organization information
org = await client.get_organization()
print(f"Organization: {org['name']}")

# Update organization settings
updated_org = await client.update_organization({
    'name': 'Acme Corp',
    'settings': {
        'enableAdvancedAnalytics': True,
        'complianceFrameworks': ['gdpr', 'hipaa']
    }
})

# Get user roles and permissions
roles = await client.get_user_roles()
print(f"User role: {roles['role']}")
print(f"Permissions: {roles['permissions']}")

# Get organization members
members = await client.get_organization_members()
print(f"{len(members)} members in organization")

# Invite new user
invitation = await client.invite_user({
    'email': '[e-mail]',
    'role': 'manager'
})

# Configure SSO
sso_config = await client.configure_sso({
    'provider': 'okta',
    'domain': 'company.okta.com',
    'clientId': 'your-client-id',
    'clientSecret': 'your-client-secret'
})

Advanced Analytics

# Compliance analytics
compliance_data = await client.get_compliance_analytics({
    'framework': 'gdpr',
    'startDate': '2024-01-01',
    'endDate': '2024-12-31'
})

# Threat detection analytics
threats = await client.get_threat_analytics({
    'severity': 'high',
    'timeframe': '24h'
})

# Performance analytics
performance = await client.get_performance_analytics({
    'metric': 'response_time',
    'aggregation': 'avg'
})

# Cost analytics
costs = await client.get_cost_analytics({
    'provider': 'openai',
    'groupBy': 'model'
})

Compliance Reporting

# Get compliance status
status = await client.get_compliance_status('gdpr')
print(f"GDPR compliance: {status['score']}%")

# Generate compliance report
report = await client.generate_compliance_report({
    'framework': 'hipaa',
    'format': 'pdf',
    'period': 'quarterly'
})

# Get compliance violations
violations = await client.get_compliance_violations({
    'severity': 'high',
    'status': 'open'
})

Policy Management

# Get all policies
policies = await client.get_policies()

# Create new policy
new_policy = await client.create_policy({
    'name': 'Content Safety Policy',
    'description': 'Prevent harmful content generation',
    'policyType': 'rego',
    'policyContent': 'package content.safety\n\ndefault allow = false...',
    'enabled': True,
    'priority': 1
})

# Update policy
updated_policy = await client.update_policy('policy-id', {
    'enabled': False,
    'priority': 2
})

# Delete policy
deleted = await client.delete_policy('policy-id')

# Test policy
test_result = await client.test_policy({
    'policyId': 'policy-id',
    'testInput': {
        'content': 'Test content',
        'user': {'role': 'user'}
    }
})

Provider Management

# Get available providers
providers = await client.get_providers()

# Add new provider
new_provider = await client.add_provider({
    'name': 'OpenAI Production',
    'providerType': 'openai',
    'apiKey': 'sk-...',
    'modelConfigs': {
        'gpt-4': {
            'maxTokens': 4000,
            'temperature': 0.7,
            'costPerToken': 0.00003
        }
    },
    'rateLimits': {
        'requestsPerMinute': 3500,
        'tokensPerMinute': 90000
    }
})

# Update provider
updated_provider = await client.update_provider('provider-id', {
    'rateLimits': {
        'requestsPerMinute': 5000
    }
})

Security & Monitoring

# Get security threats
threats = await client.get_security_threats({
    'severity': 'high',
    'timeframe': '1h'
})

# Get observability metrics
metrics = await client.get_observability_metrics({
    'metric': 'request_rate',
    'timeframe': '5m'
})

# Get system alerts
alerts = await client.get_alerts({
    'status': 'active',
    'severity': 'critical'
})

Integrations

# Get available integrations
integrations = await client.get_integrations()

# Configure integration
configured_integration = await client.configure_integration({
    'type': 'slack',
    'config': {
        'webhookUrl': 'https://hooks.slack.com/...',
        'channel': '#ai-governance'
    }
})

Data Validation

# Validate structured data
validation = await client.validate_structured_data({
    'schema': {
        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'age': {'type': 'number'}
        }
    },
    'data': {'name': 'John', 'age': 30}
})

# Validate PII detection
pii_validation = await client.validate_pii({
    'content': 'My email is [e-mail]'
})
print(f"PII detected: {pii_validation['piiDetected']}")

Audit Logging

# Get audit logs
audit_logs = await client.get_audit_logs({
    'action': 'policy_violation',
    'startDate': '2024-01-01',
    'endDate': '2024-01-31',
    'userId': 'user123'
})

print(f"Found {len(audit_logs)} audit entries")

Error Handling

The SDK provides specific exception types for different error scenarios:

from doopal_client import DoopalError, PolicyViolationError, RedactionError

try:
    response = await client.chat_completion(
        provider="openai",
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "Sensitive content here"}]
    )
except PolicyViolationError as e:
    print(f"Policy violation: {e}")
except RedactionError as e:
    print(f"Content blocked: {e}")
except DoopalError as e:
    print(f"API error: {e}")

Advanced Usage

Custom Metadata

response = await client.chat_completion(
    provider="openai",
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello"}],
    metadata={
        "user_id": "user123",
        "session_id": "session456",
        "department": "engineering"
    }
)

Multiple Providers

# OpenAI
openai_response = await client.chat_completion(
    provider="openai",
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello"}]
)

# Anthropic
anthropic_response = await client.chat_completion(
    provider="anthropic",
    model="claude-3-sonnet-20240229",
    messages=[{"role": "user", "content": "Hello"}]
)

# Azure OpenAI
azure_response = await client.chat_completion(
    provider="azure_openai",
    model="gpt-35-turbo",
    messages=[{"role": "user", "content": "Hello"}]
)

Response Format

All responses include governance and redaction information:

{
    "response": "The generated text response",
    "provider": "openai",
    "model": "gpt-3.5-turbo",
    "usage": {
        "prompt_tokens": 10,
        "completion_tokens": 20,
        "total_tokens": 30
    },
    "metadata": {
        "request_id": "req_123",
        "processing_time": 1.5,
        "policies_applied": ["content_filter", "rate_limit"],
        "cost": 0.0001
    },
    "redacted_content": [
        "[e-mail] was redacted",
        "SSN 123-45-6789 was blocked"
    ]
}

Enterprise Integration Examples

Multi-Tenant Setup

# Initialize client with organization context
client = DoopalClient(
    api_key='your-api-key',
    base_url='https://api.your-domain.com',
    organization_id='org-123'  # Important for multi-tenant deployments
)

# All subsequent calls will be scoped to this organization
org_data = await client.get_organization()
compliance = await client.get_compliance_status('gdpr')

Production Monitoring

import asyncio
from datetime import datetime

# Health monitoring for production deployments
health_checks = {
    'basic': await client.health_check(),
    'detailed': await client.get_detailed_health(),
    'ready': await client.get_readiness_status(),
    'live': await client.get_liveness_status()
}

print('Production health status:', health_checks)

# Real-time threat monitoring
async def monitor_threats():
    while True:
        threats = await client.get_security_threats({'timeframe': '5m'})
        if threats:
            print(f"{len(threats)} security threats detected!")
        await asyncio.sleep(30)  # Check every 30 seconds

# Run monitoring in background
asyncio.create_task(monitor_threats())

Compliance Automation

# Automated compliance reporting
compliance_report = await client.generate_compliance_report({
    'framework': 'sox',
    'format': 'json',
    'period': 'monthly',
    'includeViolations': True,
    'includeRemediation': True
})

# Send to compliance team
print(f"Generated SOX report: {compliance_report['reportId']}")

Enterprise Policy Management

# Bulk policy management for enterprise deployments
policies_config = [
    {
        'name': 'GDPR Data Protection',
        'description': 'Ensure GDPR compliance for EU data',
        'policyType': 'rego',
        'policyContent': open('policies/gdpr.rego').read(),
        'enabled': True,
        'priority': 1
    },
    {
        'name': 'HIPAA Healthcare Data',
        'description': 'Protect healthcare information',
        'policyType': 'rego', 
        'policyContent': open('policies/hipaa.rego').read(),
        'enabled': True,
        'priority': 2
    }
]

# Deploy policies
for policy_config in policies_config:
    policy = await client.create_policy(policy_config)
    print(f"Created policy: {policy['id']}")
    
    # Test policy
    test_result = await client.test_policy({
        'policyId': policy['id'],
        'testInput': {'content': 'Test healthcare data: Patient ID 12345'}
    })
    print(f"Policy test result: {test_result['allowed']}")

Development

Running Tests

pip install -e ".[dev]"
pytest

Code Formatting

black .
isort .

Type Checking

mypy .

Changelog

v1.1.0 (Latest)

• ✨ Added enterprise organization management
• ✨ Implemented RBAC and user management
• ✨ Added advanced compliance analytics
• ✨ Comprehensive health checks for Kubernetes
• ✨ Security threat detection and monitoring
• ✨ Policy management and testing capabilities
• ✨ Provider management and configuration
• ✨ Audit logging and compliance reporting
• ✨ Integration management
• ✨ Structured data and PII validation
• 🔧 Updated to support all B2B enterprise features
• 🔧 Enhanced async/sync method coverage

v1.0.0

• 🎉 Initial release
• ✨ Basic chat completions and embeddings
• ✨ Multi-provider support
• ✨ Content redaction and policy enforcement
• ✨ Streaming support
• ✨ OpenAI-compatible endpoints
• ✨ Async/sync interfaces

Support

• Documentation: https://docs.doopal.com
• GitHub Issues: https://github.com/doopal-ai/doopal/issues
• Email: [e-mail]
• Enterprise Support: [e-mail]
• Customer Success: [e-mail]

License

MIT License - see LICENSE file for details.