SovereignEG Python SDK

Official Python client for the SovereignEG sovereign AI platform.

Install

pip install sovereigneg

Quick Start

from sovereigneg import SovereignEG

client = SovereignEG(
    api_key="sk-seg-your-key-here",
    base_url="http://localhost:8000",  # or https://api.sovereigneg.ai
)

# Chat
response = client.chat("What is sovereign AI?")
print(response.content)

# Streaming
for token in client.chat("Tell me about Egypt", stream=True):
    print(token, end="", flush=True)

# Embeddings
emb = client.embed("Hello world")
print(f"Dimension: {len(emb.embedding)}")

Examples

See the examples/ folder:

• chat.py — basic chat completion
• streaming.py — real-time token streaming
• embeddings.py — generate embeddings
• rag.py — upload documents and search

Rate Limiting

All API responses include rate limit headers:

X-RateLimit-Limit: 60
X-RateLimit-Remaining: 42
X-RateLimit-Reset: 1771234567

When rate limited (HTTP 429), the response includes a Retry-After header and structured error:

{
  "error": {
    "type": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Your developer plan allows 60 requests per minute.",
    "plan": "developer",
    "limit": 60,
    "retry_after": 3.2
  }
}

Handling rate limits in your code:

import time
from sovereigneg import SovereignEG

client = SovereignEG(api_key="sk-seg-...")

try:
    response = client.chat("Hello!")
except Exception as e:
    if hasattr(e, 'status_code') and e.status_code == 429:
        retry_after = e.response.headers.get("Retry-After", "1")
        time.sleep(float(retry_after))
        response = client.chat("Hello!")  # Retry

Per-Plan Limits

Automatic Retries

The SDK automatically retries on transient errors (429 rate limits, 5xx server errors, network timeouts) with exponential backoff:

# Default: 2 retries. Customize:
client = SovereignEG(api_key="sk-seg-...", max_retries=5)

The SDK respects Retry-After headers from the server, so rate-limited requests wait the appropriate duration before retrying.

API Reference

SovereignEG(api_key, base_url, timeout, tenant_id, max_retries)

Create a client instance. max_retries defaults to 2.

client.chat(message, model, system, temperature, max_tokens, stream)

Send a chat completion. Returns ChatCompletion or ChatStream.

client.embed(input, model)

Generate embeddings. Returns Embedding with .embedding or .embeddings.

client.upload(file_path, tenant_id)

Upload a document for RAG. Returns DocumentUpload.

Document ACL — Control who can access uploaded documents:

# Default: all tenant members can read
client.upload("report.pdf", tenant_id="your-tenant-id")

# Group-restricted: only engineering and finance groups
client.upload("report.pdf", tenant_id="your-tenant-id",
              acl_level="group", acl_groups="engineering,finance")

# Private: only the uploader can read
client.upload("report.pdf", tenant_id="your-tenant-id", acl_level="private")

ACL levels:

• tenant (default) — all members of the tenant can read
• group — only members of the specified groups can read
• private — only the document owner (uploader) can read

ACL is enforced on search, list, and delete. Admins can always access all documents.

client.search(query, top_k, tenant_id)

Search documents. Returns list of SearchResult. Results are ACL-filtered — only documents the caller has permission to see are returned.