Python SDK
The official NEXXUS Python SDK provides an idiomatic Python interface for AI trust and safety evaluation with full async support and type hints.
Requirements
Python 3.8 or higher
Installation
pipbash
pip install trustscanpoetrybash
poetry add trustscanpipenvbash
pipenv install trustscanQuick Start
Get started with a simple evaluation in just a few lines of code:
Basic Usagepython
from trustscan import NEXXUS
# Initialize the client
client = NEXXUS(api_key="ts_live_your_api_key")
# Run an evaluation
result = client.evaluate(
content="The recommended dosage is 500mg every 6 hours.",
context="User asked about medication dosage",
mode="STANDARD"
)
# Access results
print(f"Trust Score: {result.trust_score}/100")
print(f"Verdict: {result.verdict}")
print(f"Certification: {result.certification_tier}")
# Check for issues
if result.blockers:
print("\nBlockers found:")
for blocker in result.blockers:
print(f" - {blocker.message} ({blocker.severity})")
# Review evaluator results
for evaluator in result.evaluator_results:
print(f"\n{evaluator.name}: {evaluator.verdict} ({evaluator.score}/100)")
for issue in evaluator.issues:
print(f" - {issue.message}")Authentication
Configure authentication using one of these methods:
Option 1: Direct API Key
from trustscan import NEXXUS
client = NEXXUS(api_key="ts_live_your_api_key")Option 2: Environment Variable
export TRUSTSCAN_API_KEY="ts_live_your_api_key"from trustscan import NEXXUS
# Automatically uses TRUSTSCAN_API_KEY
client = NEXXUS()Option 3: With Organization
from trustscan import NEXXUS
client = NEXXUS(
api_key="ts_live_your_api_key",
organization_id="org_abc123"
)Evaluation Modes
Choose the evaluation mode based on your needs:
| Mode | Speed | Depth | Use Case |
|---|---|---|---|
INSTANT | <100ms | Basic | Real-time filtering |
STANDARD | 1-3s | Comprehensive | Production evaluation |
DEEP | 5-30s | Exhaustive | Compliance audits |
Mode Examplespython
# Quick check for content moderation
instant = client.evaluate(content="...", mode="INSTANT")
# Standard production evaluation
standard = client.evaluate(content="...", mode="STANDARD")
# Deep analysis for compliance
deep = client.evaluate(content="...", mode="DEEP")Async Support
Use the async client for non-blocking operations:
Async Clientpython
import asyncio
from trustscan import AsyncNEXXUS
async def main():
client = AsyncNEXXUS(api_key="ts_live_your_api_key")
# Run evaluation asynchronously
result = await client.evaluate(
content="AI-generated response",
mode="STANDARD"
)
print(f"Trust Score: {result.trust_score}")
# Close the client when done
await client.close()
asyncio.run(main())Context Managerpython
async with AsyncNEXXUS() as client:
result = await client.evaluate(content="...")Parallel Evaluationspython
async def evaluate_batch(contents: list[str]):
async with AsyncNEXXUS() as client:
# Run multiple evaluations concurrently
tasks = [
client.evaluate(content=content, mode="STANDARD")
for content in contents
]
results = await asyncio.gather(*tasks)
return results
# Evaluate 10 items concurrently
results = asyncio.run(evaluate_batch(my_contents))Long-running Evaluations
For large content or DEEP mode, use async evaluations with polling:
Async Evaluation with Pollingpython
# Start async evaluation
async_result = client.evaluate_async(
content=large_document,
mode="DEEP"
)
print(f"Evaluation ID: {async_result.evaluation_id}")
print(f"Status: {async_result.status}")
# Poll for completion
result = client.wait_for_evaluation(
evaluation_id=async_result.evaluation_id,
poll_interval=2.0, # seconds
max_wait=300.0 # 5 minutes
)
print(f"Completed! Trust Score: {result.trust_score}")With Webhookpython
# Start evaluation with webhook callback
async_result = client.evaluate_async(
content=large_document,
mode="DEEP",
webhook_url="https://your-app.com/webhooks/trustscan"
)
# Your webhook will receive the result when completeSystems Management
Track AI systems and their evaluation history:
Create and Evaluate Systempython
# Create a system
system = client.create_system(
name="Customer Support Bot",
industry="retail",
system_type="chatbot",
description="AI-powered customer support"
)
print(f"System ID: {system.system_id}")
# Evaluate the system
result = client.evaluate_system(
system_id=system.system_id,
content="Bot response to evaluate",
mode="STANDARD"
)
# Get evaluation history
history = client.get_system_history(
system_id=system.system_id,
limit=10
)
for eval in history.evaluations:
print(f"{eval.timestamp}: Score {eval.trust_score}")List and Filter Systemspython
# List all systems
systems = client.list_systems()
# Filter by criteria
healthcare_systems = client.list_systems(
industry="healthcare",
min_trust_score=80,
certification_tier="GOLD"
)
for system in healthcare_systems.systems:
print(f"{system.name}: {system.trust_score}")Error Handling
Handle errors gracefully with specific exception types:
Exception Handlingpython
from trustscan import NEXXUS
from trustscan.exceptions import (
NEXXUSError,
AuthenticationError,
RateLimitError,
ValidationError,
TimeoutError,
)
client = NEXXUS()
try:
result = client.evaluate(content="...")
except AuthenticationError:
print("Invalid API key")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ValidationError as e:
print(f"Invalid request: {e.message}")
print(f"Details: {e.details}")
except TimeoutError:
print("Request timed out")
except NEXXUSError as e:
print(f"API error: {e.message} (code: {e.code})")Configuration
Customize client behavior:
Advanced Configurationpython
from trustscan import NEXXUS
client = NEXXUS(
api_key="ts_live_your_api_key",
organization_id="org_abc123",
base_url="https://api.trustscan.io", # Custom endpoint
timeout=60.0, # Request timeout in seconds
max_retries=3, # Retry attempts for transient failures
retry_delay=1.0, # Initial retry delay
)
# Or configure via environment variables:
# TRUSTSCAN_API_KEY=ts_live_xxx
# TRUSTSCAN_BASE_URL=https://api.trustscan.io
# TRUSTSCAN_ORG_ID=org_xxx
# TRUSTSCAN_TIMEOUT=60Common Use Cases
LLM Response Validation
def validate_llm_response(prompt: str, response: str) -> bool:
"""Validate LLM response before sending to user."""
result = client.evaluate(
content=response,
context=prompt,
mode="INSTANT"
)
if result.verdict in ("FAIL", "HARD_FAIL"):
# Log or handle the failure
logger.warning(f"Response failed validation: {result.blockers}")
return False
return TrueBatch Processing
import asyncio
from trustscan import AsyncNEXXUS
async def process_batch(items: list[dict]) -> list[dict]:
"""Process a batch of items with trust evaluation."""
async with AsyncNEXXUS() as client:
tasks = []
for item in items:
task = client.evaluate(
content=item["content"],
context=item.get("context"),
mode="STANDARD"
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
{
"id": item["id"],
"trust_score": r.trust_score if not isinstance(r, Exception) else None,
"error": str(r) if isinstance(r, Exception) else None,
}
for item, r in zip(items, results)
]CI/CD Integration
#!/usr/bin/env python
"""Pre-commit hook for trust evaluation."""
import sys
from trustscan import NEXXUS
def main():
client = NEXXUS() # Uses TRUSTSCAN_API_KEY env var
# Read content to evaluate
with open("generated_responses.json") as f:
responses = json.load(f)
failed = False
for response in responses:
result = client.evaluate(
content=response["text"],
mode="STANDARD"
)
if result.verdict == "HARD_FAIL":
print(f"HARD FAIL: {response['id']}")
for blocker in result.blockers:
print(f" - {blocker.message}")
failed = True
sys.exit(1 if failed else 0)
if __name__ == "__main__":
main()Type Hints
The SDK includes full type annotations for better IDE support:
from trustscan import NEXXUS
from trustscan.types import (
EvaluationRequest,
EvaluationResponse,
EvaluationMode,
Verdict,
CertificationTier,
)
def evaluate_with_types(client: NEXXUS, content: str) -> EvaluationResponse:
result: EvaluationResponse = client.evaluate(
content=content,
mode=EvaluationMode.STANDARD
)
verdict: Verdict = result.verdict
tier: CertificationTier = result.certification_tier
return result