lumnisai 0.2.7

Agent-oriented SDK for the Lumnis multi-tenant AI platform

LumnisAI Python SDK

The official Python SDK for the LumnisAI multi-tenant AI platform. Build agent-oriented applications with support for multiple AI providers, user scoping, and conversation threads.

Features

• Multi-tenant Architecture: Scope operations to tenants or individual users
• User Management: Full CRUD operations for user accounts with cascade deletion
• Multiple AI Providers: Support for OpenAI, Anthropic, Google, and Azure
• Model Preferences: Configure preferred models for different use cases (cheap, fast, smart, reasoning, vision)
• Async & Sync APIs: Both synchronous and asynchronous client interfaces
• Conversation Threads: Maintain conversation context across interactions
• Structured Output: Get responses in JSON format using Pydantic models
• Progress Tracking: Real-time progress updates with customizable callbacks
• Type Safety: Full type hints and Pydantic models for robust development
• Error Handling: Comprehensive exception hierarchy for different error scenarios

Installation

pip install lumnisai

For development:

pip install lumnisai[dev]

Quick Start

Synchronous Client

import lumnisai

# Initialize client (defaults to user scope)
client = lumnisai.Client()

# Simple AI interaction (requires user_id in user scope)
response = client.invoke(
    "Analyze the latest trends in machine learning",
    user_id="user-123"
)

print(response.output_text)

Asynchronous Client

import asyncio
import lumnisai

async def main():
    # Auto-initializes on first use (defaults to user scope)
    client = lumnisai.AsyncClient()
    response = await client.invoke(
        "Write a summary of quantum computing advances",
        user_id="user-123"
    )
    print(response.output_text)
    
    # Optional cleanup
    await client.close()

asyncio.run(main())

Streaming Responses

async def stream_example():
    # Auto-initializes on first use (defaults to user scope)
    client = lumnisai.AsyncClient()
    async for update in await client.invoke(
        "Conduct research on renewable energy trends",
        stream=True,
        user_id="user-123"
    ):
        print(f"Status: {update.status}")
        if update.status == "succeeded":
            print(f"Final result: {update.output_text}")
    
    # Optional cleanup
    await client.close()

asyncio.run(stream_example())

Invoke API: Unified Interface

The invoke() method provides a unified interface for both blocking and streaming responses:

# Blocking response (default)
response = await client.invoke("Hello world", user_id="user-123")
print(response.output_text)

# Streaming response  
async for update in await client.invoke("Hello world", stream=True, user_id="user-123"):
    print(f"Status: {update.status}")
    if update.status == "succeeded":
        print(update.output_text)

Benefits:

• Single method - No confusion between invoke() vs invoke_stream()
• Clear parameter - stream=True makes intent obvious
• Type safety - Proper type hints for both use cases
• Backwards compatible - invoke_stream() still works (deprecated)

Structured Output

Get AI responses in structured JSON format using Pydantic models. Perfect for extracting specific data, building APIs, or integrating with other systems.

Basic Usage

from pydantic import BaseModel, Field

# Define your output structure
class ProductInfo(BaseModel):
    name: str = Field(description="Product name")
    price: float = Field(description="Price in USD")
    in_stock: bool = Field(description="Whether item is in stock")

# Pass the model directly to invoke
response = client.invoke(
    "Tell me about the iPhone 15 Pro",
    response_format=ProductInfo,  # Pass Pydantic model class
    user_id="user-123"
)

# Access structured data
if response.structured_response:
    product = ProductInfo(**response.structured_response)
    print(f"{product.name}: ${product.price} ({'In Stock' if product.in_stock else 'Out of Stock'})")

Complex Nested Structures

class Address(BaseModel):
    street: str
    city: str
    country: str

class BusinessInfo(BaseModel):
    name: str
    category: str
    address: Address
    rating: Optional[float] = Field(None, ge=0, le=5)

response = await client.invoke(
    "Tell me about Tesla's headquarters",
    response_format=BusinessInfo,
    user_id="user-123"
)

if response.structured_response:
    business = BusinessInfo(**response.structured_response)
    print(f"{business.name} ({business.category})")
    print(f"Location: {business.address.city}, {business.address.country}")

Response Format Instructions

Add specific instructions for how the structured output should be formatted:

class WeatherData(BaseModel):
    temperature: str
    conditions: str
    humidity: str

response = client.invoke(
    "What's the weather in Paris?",
    response_format=WeatherData,
    response_format_instructions="Use Celsius for temperature and include the % symbol for humidity",
    user_id="user-123"
)

Using JSON Schema Directly

You can also pass a JSON schema dictionary instead of a Pydantic model:

response = client.invoke(
    "Analyze this product review",
    response_format={
        "type": "object",
        "properties": {
            "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
            "score": {"type": "number", "minimum": 0, "maximum": 10},
            "summary": {"type": "string"}
        },
        "required": ["sentiment", "score", "summary"]
    },
    user_id="user-123"
)

Important Notes

• Both output_text and structured_response are returned in the response
• If the AI cannot generate valid structured output, structured_response may be None
• Always validate the structured response before using it
• The structured output feature works with both sync and async clients

Model Preferences

Configure which LLM models to use for different scenarios. The SDK supports five model types:

• CHEAP_MODEL: Cost-effective for simple tasks
• FAST_MODEL: Low latency for quick responses
• SMART_MODEL: High quality for complex tasks
• REASONING_MODEL: Advanced reasoning and logic
• VISION_MODEL: Image understanding capabilities

Configuring Model Preferences

# First, configure API keys for providers
client.add_api_key(provider="OPENAI_API_KEY", api_key="sk-...")
client.add_api_key(provider="ANTHROPIC_API_KEY", api_key="sk-ant-...")

# Get current preferences
preferences = client.get_model_preferences()
for pref in preferences.preferences:
    print(f"{pref.model_type}: {pref.provider}:{pref.model_name}")

# Update preferences (bulk update)
client.update_model_preferences({
    "FAST_MODEL": {"provider": "openai", "model_name": "gpt-4o-mini"},
    "SMART_MODEL": {"provider": "anthropic", "model_name": "claude-3-7-sonnet-20250219"}
})

Runtime Model Overrides

Override model selection for specific requests:

# Create response with model override
response = client.responses.create(
    messages=[
        {"role": "user", "content": "Solve this complex problem"}
    ],
    model_overrides={
        "smart_model": "anthropic:claude-3-7-sonnet-20250219"
    }
)

# Override multiple models
response = client.responses.create(
    messages=[
        {"role": "user", "content": "Analyze this data"}
    ],
    model_overrides={
        "fast_model": "openai:gpt-4o-mini",
        "smart_model": "openai:gpt-4o",
        "reasoning_model": "openai:o1"
    }
)

Configuration

Environment Variables

Set up your environment with the following variables:

export LUMNISAI_API_KEY="your-api-key"
export LUMNISAI_BASE_URL="https://api.lumnis.ai"  # Optional
export LUMNISAI_TENANT_ID="your-tenant-id"       # Optional - auto-detected from API key

Client Configuration

client = lumnisai.Client(
    api_key="your-api-key",           # Required
    base_url="https://api.lumnis.ai", # Optional
    tenant_id="your-tenant-id",       # Optional - auto-detected from API key
    timeout=30.0,                     # Request timeout
    max_retries=3,                    # Retry attempts
    scope=Scope.USER                  # Default scope
)

Note on Tenant ID: The tenant_id parameter is optional because each API key is automatically scoped to a specific tenant. The SDK will extract the tenant context from your API key. You only need to explicitly provide tenant_id if you're using a special cross-tenant API key (rare).

Understanding Scopes: Tenant vs User

LumnisAI operates in a multi-tenant architecture where each tenant can have multiple users. Understanding the difference between tenant and user scope is crucial for proper implementation.

Important: As of v0.2.0, the SDK defaults to User scope for better security and data isolation. This is a breaking change from earlier versions.

Tenant Scope vs User Scope

Aspect	Tenant Scope	User Scope
Purpose	System-wide operations for the entire organization	User-specific operations and data isolation
Data Access	Access to all tenant data	Access only to user's own data
Use Cases	Admin dashboards, analytics, system operations	End-user applications, personal assistants
Permissions	Requires admin-level API keys	Standard user API keys
user_id	❌ Must NOT be provided	✅ Required

When to Use Each Scope

Use Tenant Scope when:

• Building admin dashboards or management interfaces
• Performing system-wide analytics or reporting
• Implementing tenant-level configuration changes
• Running background jobs that affect all users
• You have admin-level permissions

Use User Scope when:

• Building end-user applications (chatbots, assistants)
• Each user should only see their own data
• Implementing user-specific features
• Building customer-facing applications
• Following principle of least privilege

User-Scoped Operations

# Method 1: Pass user_id to each call
client = lumnisai.Client(scope=Scope.USER)
response = client.invoke("Hello", user_id="user-123")

# Method 2: Create user-scoped client
user_client = client.for_user("user-123")
response = user_client.invoke("Hello")

# Method 3: Temporary user context
with client.as_user("user-123") as user_client:
    response = user_client.invoke("Hello")

# Method 4: Explicit user scope with user_id
client = lumnisai.Client(scope=Scope.USER)
response = client.invoke("Hello", user_id="user-123")

Tenant-Scoped Operations

# Use tenant scope (requires proper permissions)
client = lumnisai.Client(scope=Scope.TENANT)

# System-wide queries (no user_id needed)
response = client.invoke("Generate monthly usage report")

# List all users' responses
all_responses = client.list_responses()

# Access tenant-level settings
tenant_info = client.tenant.get()

Scope Validation and Error Handling

The SDK automatically validates scope usage and provides clear error messages:

import lumnisai
from lumnisai.exceptions import MissingUserId, TenantScopeUserIdConflict

# ❌ This will raise MissingUserId
try:
    client = lumnisai.Client(scope=Scope.USER)
    response = client.invoke("Hello")  # Missing user_id
except MissingUserId:
    print("user_id is required when scope is USER")

# ❌ This will raise TenantScopeUserIdConflict  
try:
    client = lumnisai.Client(scope=Scope.TENANT)
    response = client.invoke("Hello", user_id="user-123")  # user_id not allowed
except TenantScopeUserIdConflict:
    print("user_id must not be provided when scope is TENANT")

User Management

Manage users within your tenant with full CRUD operations:

# Create a new user
user = await client.create_user(
    email="[e-mail]",
    first_name="Alice",
    last_name="Johnson"
)

# Get user by ID or email
user = await client.get_user("550e8400-e29b-41d4-a716-446655440000")
user = await client.get_user("[e-mail]")

# Update user information
updated_user = await client.update_user(
    user.id,
    first_name="Alicia",
    last_name="Smith"
)

# List all users with pagination
users_response = await client.list_users(page=1, page_size=20)
for user in users_response.users:
    print(f"{user.email} - {user.first_name} {user.last_name}")

# Delete user (cascades to all user data)
await client.delete_user(user.id)

Synchronous User Management

# Works the same with sync client
client = lumnisai.Client()

user = client.create_user(
    email="[e-mail]",
    first_name="Bob",
    last_name="Wilson"
)

users = client.list_users(page_size=50)
print(f"Total users: {users.pagination.total}")

Conversation Threads

# Create a new thread
thread = client.create_thread(
    user_id="user-123",
    title="Research Project"
)

# Continue conversation in thread
response1 = client.invoke(
    "What is machine learning?",
    user_id="user-123",
    thread_id=thread.thread_id
)

response2 = client.invoke(
    "Can you give me specific examples?",
    user_id="user-123", 
    thread_id=thread.thread_id
)

# List user's threads
threads = client.list_threads(user_id="user-123")

Progress Tracking

Enable automatic progress printing with the show_progress=True parameter:

# Automatic progress tracking (prints status and message updates)
response = await client.invoke(
    "Research the latest AI developments and write a report",
    user_id="user-123",
    show_progress=True  # Prints status changes and progress messages
)

# Output example:
# Status: in_progress
# PLANNING: Starting research on AI developments
# RESEARCHING: Gathering information from recent sources
# WRITING: Composing comprehensive report
# Status: succeeded

Benefits:

• Simple - Just add show_progress=True
• Automatic - No custom callbacks needed
• Clean output - Only prints when status or messages change
• Works everywhere - Both sync and async clients

API Key Management (External API Keys)

Configure API keys for different AI providers to use their models:

Supported Providers

All available API key providers:

• OPENAI_API_KEY - OpenAI models (GPT-4, etc.)
• ANTHROPIC_API_KEY - Anthropic Claude models
• GOOGLE_API_KEY - Google Gemini models
• COHERE_API_KEY - Cohere models
• GROQ_API_KEY - Groq cloud models
• NVIDIA_API_KEY - NVIDIA models
• FIREWORKS_API_KEY - Fireworks AI models
• MISTRAL_API_KEY - Mistral AI models
• TOGETHER_API_KEY - Together AI models
• XAI_API_KEY - xAI Grok models
• PPLX_API_KEY - Perplexity models
• HUGGINGFACE_API_KEY - Hugging Face models
• DEEPSEEK_API_KEY - DeepSeek models
• IBM_API_KEY - IBM models

Managing API Keys

# Add API keys
client.add_api_key(
    provider="OPENAI_API_KEY",
    api_key="sk-..."
)

client.add_api_key(
    provider="ANTHROPIC_API_KEY",
    api_key="sk-ant-..."
)

# List your API keys
keys = client.list_api_keys()
for key in keys:
    print(f"Provider: {key.provider}, Active: {key.is_active}")

# Delete an API key
client.delete_api_key("OPENAI_API_KEY")

Error Handling

import lumnisai
from lumnisai.exceptions import (
    AuthenticationError,
    MissingUserId,
    TenantScopeUserIdConflict,
    ValidationError,
    RateLimitError,
    NotFoundError
)

try:
    response = client.invoke("Hello", user_id="user-123")
except AuthenticationError:
    print("Invalid API key")
except MissingUserId:
    print("User ID required for user-scoped operations")
except TenantScopeUserIdConflict:
    print("Cannot specify user_id with tenant scope")
except ValidationError as e:
    print(f"Invalid request: {e}")
except RateLimitError:
    print("Rate limit exceeded")
except NotFoundError:
    print("Resource not found")

Advanced Usage

Message Format

# String format (converted to user message)
response = client.invoke("Hello world", user_id="user-123")

# Single message object
response = client.invoke(
    {"role": "user", "content": "Hello world"},
    user_id="user-123"
)

# Multiple messages (conversation history)
response = client.invoke([
    {"role": "user", "content": "What is Python?"},
    {"role": "assistant", "content": "Python is a programming language..."},
    {"role": "user", "content": "Give me an example"}
], user_id="user-123")

Response Management

# Create response without waiting
response = await client.responses.create(
    messages=[{"role": "user", "content": "Hello"}],
    user_id="user-123"
)

# Poll for completion manually
final_response = await client.get_response(
    response.response_id,
    wait=30.0  # Wait up to 30 seconds
)

# Cancel a response
cancelled = await client.cancel_response(response.response_id)

# List user's responses
responses = client.list_responses(user_id="user-123", limit=10)

Idempotency

# Ensure exactly-once processing
response = client.invoke(
    "Important calculation",
    user_id="user-123",
    idempotency_key="calc-2024-001"
)

# Subsequent calls with same key return original response
duplicate = client.invoke(
    "Important calculation", 
    user_id="user-123",
    idempotency_key="calc-2024-001"  # Same key
)

assert response.response_id == duplicate.response_id

Development

Installation

git clone https://github.com/lumnisai/lumnisai-python.git
cd lumnisai-python

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install dependencies and package in development mode
uv sync --dev
uv pip install -e .

Running Tests

# Run all tests
pytest

# Run specific test files
python local_dev/test_01_auth.py
python local_dev/test_02_basic.py

# Run with coverage
pytest --cov=lumnisai

Code Quality

# Formatting
black .
isort .

# Linting
ruff check .

# Type checking
mypy lumnisai/

API Reference

Core Classes

• Client: Synchronous client for LumnisAI API
• AsyncClient: Asynchronous client for LumnisAI API
• ResponseObject: Represents an AI response with progress tracking
• ThreadObject: Represents a conversation thread

Enums

• Scope: USER or TENANT - defines operation scope
• ApiProvider: OPENAI, ANTHROPIC, GOOGLE, AZURE
• ApiKeyMode: BRING_YOUR_OWN, USE_PLATFORM

Resources

• responses: Manage AI responses
• threads: Manage conversation threads
• external_api_keys: Manage external provider API keys
• tenant: Tenant-level operations
• users: User management (CRUD operations)

Support

• Documentation: https://lumnisai.github.io/lumnisai-python
• Issues: https://github.com/lumnisai/lumnisai-python/issues
• Email: [e-mail]

License

This project is licensed under the Apache 2.0 License - see the LICENSE file for details.

Contributing

We welcome contributions! Please see our Contributing Guide for details on:

• Setting up the development environment
• Running tests and code quality checks
• Submitting pull requests
• Code style guidelines

---

Built with ❤️ by the LumnisAI team