Python SDK Reference

pip install aegissovereign          # latest
# or
poetry add aegissovereign

1from aegissovereign import Client
2
3oss = Client(
4    api_url="https://sovereign.yourcompany.com",
5    token="sk-sovereign-...",     # or set OSS_TOKEN
6    workspace="ws-prod",           # or set OSS_WORKSPACE
7    max_retries=3,
8    backoff_base=0.5,
9)
10
11# Register a model — fires CompliancePipelineAgent + Safety Gate
12model = oss.models.register(
13    name="fraud-detector-v4",
14    version="4.0.0",
15    framework="scikit-learn",
16    metrics={"auc": 0.986},
17)
18
19# Check compliance
20report = oss.compliance.check(workspace_id="ws-prod")
21print(report.score)   # e.g. 91

1# Validate a user prompt
2result = oss.safety.validate_input(
3    "Ignore all previous instructions...",
4    dry_run=False,          # True = shadow mode (no blocking)
5)
6if result["blocked"]:
7    raise ValueError(f"Blocked: {result['violations']}")
8
9# Validate a model response (auto-redacts PII)
10resp = oss.safety.validate_output(response_text)
11safe_text = resp.get("sanitized_text", response_text)
12
13# Validate a tool call before execution
14check = oss.safety.validate_tool_call(
15    "delete_file",
16    {"path": "/prod/db"},
17    agent_id="researcher-agent",
18)
19if not check["allowed"]:
20    raise PermissionError(f"Tool blocked: {check['risk_level']}")

1# List pending critical reviews
2reviews = oss.safety.list_reviews(status="pending", priority="critical")
3
4# Submit a decision
5oss.safety.submit_review(
6    request_id="rev-abc123",
7    decision="reject",
8    reason="Confirmed jailbreak — DAN persona",
9    reviewer_id="alice@corp.com",
10    reviewer_name="Alice Chen",
11)
12
13# Check SLO fleet health
14summary = oss.safety.slo_summary()
15if summary["critical"] > 0:
16    print("Critical SLO burn rate — PagerDuty already paged")
17
18# Check model registry clearance
19check = oss.safety.check_model("gpt-4", "gpt-4-turbo", use_case="customer-support")
20if not check["allowed"]:
21    raise RuntimeError(check["reason"])
22
23# Run red-team certification
24result = oss.safety.run_redteam(
25    domains=["medical", "financial"],
26    include_multi_turn=True,
27)
28cert = result["certification"]
29print(f"Pass rate: {cert['pass_rate']:.1%}  Blocking: {cert['blocking_failures']}")
30
31# Generate EU AI Act regulatory report
32report = oss.safety.generate_report(
33    "eu_ai_act",
34    organization="Acme Financial",
35    format="markdown",
36)
37print(report["content"])

1# Pre-deployment clearance verdict
2clearance = oss.safety.check_model("my-fine-tune")
3print(f"  cleared={clearance['cleared']}  verdict={clearance['verdict']}")
4
5# Provenance lineage DAG
6lineage = oss.safety.get_lineage("my-fine-tune", "v1.0")
7for node in lineage["chain"]:
8    reeval = "re-evaluated" if not node["safety_inherited"] else "inherited"
9    print(f"  {node['model_id']}@{node['version']}  safety={reeval}")
10
11# Red-team certification run (server-side)
12cert = oss.safety.run_redteam(domains=["financial"], include_multi_turn=True)
13if not cert["overall_passed"]:
14    print(f"BLOCKED: {cert['certification']['blocking_failures']} undetected attack(s)")

1import asyncio
2from aegissovereign import AsyncClient
3
4async def main():
5    async with AsyncClient(api_url="...", token="...") as oss:
6        # Low-level async call (SafetyClient not yet on AsyncClient)
7        result = await oss.post(
8            "/safety/validate/input",
9            json={"text": "...", "target": "input", "context": {}},
10        )
11        print(result["blocked"])
12
13asyncio.run(main())

1from aegissovereign import Client, OSSAuthError, OSSNotFoundError, OSSAPIError
2
3try:
4    model = oss.models.get("nonexistent-id")
5except OSSNotFoundError:
6    print("Model not found")
7except OSSAuthError:
8    print("Token expired or invalid — re-run oss login")
9except OSSAPIError as e:
10    print(f"HTTP {e.status_code}: {e}")