sensoit 1.2.0

Official Python SDK for Sensoit - AI Prompt Security & Management

Sensoit Python SDK

Official Python SDK for Sensoit - AI Prompt Security & Management Platform.

Sensoit provides enterprise-grade guardrails for AI applications, including PII detection, toxicity filtering, keyword blocking, and custom policy enforcement.

Installation

pip install sensoit-sdk

# With framework integrations
pip install sensoit-sdk[fastapi]
pip install sensoit-sdk[flask]
pip install sensoit-sdk[django]
pip install sensoit-sdk[all]  # All frameworks

Quick Start

Async Usage

from sensoit import Sensoit

async def main():
    async with Sensoit(api_key="gf_live_xxx") as gf:
        result = await gf.run(
            "customer-support",
            input="I need help with my order #12345",
            variables={"customer_name": "John"},
            session_id="session_abc",
        )

        if result.blocked:
            print(f"Blocked: {result.violations}")
        else:
            print(f"Response: {result.output}")

import asyncio
asyncio.run(main())

Sync Usage

from sensoit import Sensoit

gf = Sensoit(api_key="gf_live_xxx")

result = gf.run_sync(
    "customer-support",
    input="I need help with my order #12345",
    variables={"customer_name": "John"},
)

print(f"Response: {result.output}")
print(f"Tokens: {result.tokens_used}")
print(f"Cost: ${result.cost_usd}")

Configuration

gf = Sensoit(
    # Required: Your API key
    api_key="gf_live_xxx",

    # Optional: API base URL (default: https://api.sensoit.io)
    base_url="https://api.sensoit.io",

    # Optional: Request timeout in seconds (default: 30.0)
    timeout=30.0,

    # Optional: Max retry attempts (default: 3)
    max_retries=3,

    # Optional: Enable debug logging (default: False)
    debug=True,

    # Optional: Enable caching (default: True)
    cache_enabled=True,

    # Optional: Cache TTL in seconds (default: 60)
    cache_ttl_seconds=60,
)

Option	Type	Default	Description
api_key	str	required	Your Sensoit API key
base_url	str	https://api.sensoit.io	API base URL
timeout	float	30.0	Request timeout in seconds
max_retries	int	3	Max retry attempts
debug	bool	False	Enable debug logging
cache_enabled	bool	True	Enable prompt caching
cache_ttl_seconds	int	60	Cache TTL in seconds

Core Methods

run() / run_sync()

Run a prompt through Sensoit with guardrail protection.

# Async
result = await gf.run(
    "support-assistant",
    input="I need help with my order",
    variables={"customer_name": "John"},
    session_id="session_123",
    force_refresh=False,
    dry_run=False,
    timeout=60.0,
    metadata={"source": "web"},
)

# Sync
result = gf.run_sync("support-assistant", input="Hello")

Return Type: RunResult

@dataclass
class RunResult:
    output: str           # Generated response
    allowed: bool         # Request passed guardrails
    blocked: bool         # Request was blocked
    escalated: bool       # Request needs human review
    violations: list      # Policy violations
    tokens_used: int      # Total tokens (in + out)
    tokens_in: int        # Input tokens
    tokens_out: int       # Output tokens
    latency_ms: int       # Total latency
    cost_usd: float       # Estimated cost
    prompt_version: int   # Prompt version used
    model: str            # LLM model used
    provider: str         # LLM provider
    cached: bool          # Served from cache
    request_id: str       # Unique request ID

run_batch() / run_batch_sync()

Run multiple prompts in parallel.

# Async
batch_result = await gf.run_batch([
    {"prompt": "translator", "options": {"input": "Hello", "variables": {"lang": "es"}}},
    {"prompt": "translator", "options": {"input": "Goodbye", "variables": {"lang": "fr"}}},
])

# Sync
batch_result = gf.run_batch_sync([...])

print(f"Total: {batch_result.summary.total}")
print(f"Allowed: {batch_result.summary.allowed}")
print(f"Blocked: {batch_result.summary.blocked}")

for result in batch_result.results:
    print(f"{result.prompt}: {result.output}")

invalidate_cache()

Clear cached prompt data.

# Clear cache for specific prompt
gf.invalidate_cache("support-assistant")

# Clear all cache
gf.invalidate_cache()

FastAPI Integration

from fastapi import FastAPI, Request
from sensoit import Sensoit
from sensoit.middleware import SensoitMiddleware

app = FastAPI()
gf = Sensoit(api_key="gf_live_xxx")

# Add middleware to protect routes
app.add_middleware(
    SensoitMiddleware,
    client=gf,
    prompt="support-bot",
    protected_paths=["/api/chat", "/api/ask"],
    blocked_response={"error": "Cannot help with that request"},
    blocked_status_code=200,
)

@app.post("/api/chat")
async def chat(request: Request):
    # Access Sensoit result
    result = request.scope.get("sensoit")
    return {"reply": result.output if result else "No response"}

Flask Integration

from flask import Flask, request, jsonify
from sensoit import Sensoit
from sensoit.middleware import flask_protect

app = Flask(__name__)
gf = Sensoit(api_key="gf_live_xxx")

@app.route("/chat", methods=["POST"])
@flask_protect(gf, prompt="support-bot")
def chat():
    # Access Sensoit result
    result = request.sensoit
    return jsonify({"reply": result.output})

# With custom options
@app.route("/ask", methods=["POST"])
@flask_protect(
    gf,
    prompt="qa-bot",
    extract_input=lambda: request.json.get("question"),
    extract_variables=lambda: {"user_id": request.json.get("user_id")},
    blocked_response={"error": "Question not allowed"},
)
def ask():
    return jsonify({"answer": request.sensoit.output})

Django Integration

# settings.py
MIDDLEWARE = [
    # ... other middleware
    'sensoit.middleware.DjangoSensoitMiddleware',
]

SENSOIT = {
    'API_KEY': 'gf_live_xxx',
    'PROMPT': 'support-bot',
    'PROTECTED_PATHS': ['/api/chat/', '/api/ask/'],
    'BASE_URL': 'https://api.sensoit.io',  # Optional
}

# views.py
from django.http import JsonResponse

def chat(request):
    # Access Sensoit result
    result = getattr(request, "sensoit", None)
    if result:
        return JsonResponse({"reply": result.output})
    return JsonResponse({"error": "No response"})

Error Handling

from sensoit import (
    Sensoit,
    BlockedError,
    EscalatedError,
    RateLimitError,
    AuthError,
    TimeoutError,
    ValidationError,
    is_retryable_error,
)

gf = Sensoit(api_key="gf_live_xxx")

try:
    result = await gf.run("my-prompt", input="Hello")
    print(result.output)

except BlockedError as e:
    print(f"Blocked by: {[v['policyName'] for v in e.violations]}")
    print(f"Has critical: {e.has_critical_violation()}")

except EscalatedError as e:
    print(f"Escalated: {e.reason}")

except RateLimitError as e:
    print(f"Rate limited, retry after: {e.retry_after_seconds}s")

except AuthError as e:
    print(f"Auth error: {e.message}")

except TimeoutError as e:
    print(f"Timed out after: {e.timeout_seconds}s")

except ValidationError as e:
    print(f"Validation errors: {e.field_errors}")

Error Classes

Error	Code	Status	Description
BlockedError	BLOCKED	200	Response blocked by guardrails
EscalatedError	ESCALATED	200	Response needs human review
RateLimitError	RATE_LIMIT	429	Rate limit exceeded
AuthError	AUTH_ERROR	401	Invalid API key
TimeoutError	TIMEOUT	-	Request timed out
ValidationError	VALIDATION_ERROR	400	Invalid input
PromptNotFoundError	PROMPT_NOT_FOUND	404	Prompt doesn't exist
NetworkError	NETWORK_ERROR	-	Network failure
SensoitAPIError	API_ERROR	varies	General API error

Event Callbacks

gf = Sensoit(api_key="gf_live_xxx")

# Register event handlers
@gf.on_blocked
def handle_blocked(prompt_name, violations, request_id):
    print(f"Blocked: {prompt_name}")
    for v in violations:
        print(f"  - {v['policyName']}: {v['detail']}")

@gf.on_escalated
def handle_escalated(prompt_name, request_id, reason):
    print(f"Escalated: {prompt_name} - {reason}")

@gf.on_error
def handle_error(prompt_name, error, request_id):
    print(f"Error in {prompt_name}: {error}")

@gf.on_complete
def handle_complete(prompt_name, result):
    print(f"Completed: {prompt_name}")
    print(f"  Tokens: {result.tokens_used}")
    print(f"  Cost: ${result.cost_usd}")

@gf.on_retry
def handle_retry(prompt_name, attempt, max_attempts):
    print(f"Retry {attempt}/{max_attempts} for {prompt_name}")

Type Hints

All types are exported for type checking:

from sensoit import (
    # Configuration
    SensoitConfig,
    CacheConfig,

    # Run operations
    RunOptions,
    RunResult,

    # Batch operations
    BatchRunInput,
    BatchRunResult,
    BatchRunItemResult,
    BatchRunSummary,

    # Guardrails
    GuardrailViolation,
    GuardrailViolationType,
    ViolationSeverity,
    ViolationAction,
)

Caching

The SDK includes built-in caching for prompt responses:

gf = Sensoit(
    api_key="gf_live_xxx",
    cache_enabled=True,
    cache_ttl_seconds=60,
)

# First call - makes API request
result1 = await gf.run("prompt", input="Hello")
print(result1.cached)  # False

# Second call - served from cache
result2 = await gf.run("prompt", input="Hello")
print(result2.cached)  # True

# Force fresh request
result3 = await gf.run("prompt", input="Hello", force_refresh=True)
print(result3.cached)  # False

# Invalidate cache
gf.invalidate_cache("prompt")

Health Check

# Async
health = await gf.health()

# Sync
health = gf.health_sync()

print(health)
# {'status': 'ok', 'service': 'sensoit', 'timestamp': '...'}

Context Manager

# Async
async with Sensoit(api_key="gf_live_xxx") as gf:
    result = await gf.run("prompt", input="Hello")

# Sync
with Sensoit(api_key="gf_live_xxx") as gf:
    result = gf.run_sync("prompt", input="Hello")

Environment Variables

export SENSOIT_API_KEY=gf_live_xxx
export SENSOIT_BASE_URL=https://api.sensoit.io

import os
from sensoit import Sensoit

gf = Sensoit(
    api_key=os.environ["SENSOIT_API_KEY"],
    base_url=os.environ.get("SENSOIT_BASE_URL", "https://api.sensoit.io"),
)

Requirements

• Python 3.9 or higher
• httpx >= 0.26.0
• pydantic >= 2.0.0
• tenacity >= 8.0.0

License

MIT

Support

• Documentation: https://docs.sensoit.io
• Issues: https://github.com/sensoit/sensoit-sdk/issues
• Email: [e-mail]