User Management - Python SDK

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Create or update user
user = client.create_user(
    email="user@example.com",
    first_name="John",
    last_name="Doe"
)
 
print(f"User ID: {user.id}")
print(f"Email: {user.email}")

# Get user by email
user = client.get_user("user@example.com")
 
# Get user by ID
user = client.get_user("550e8400-e29b-41d4-a716-446655440000")
 
print(f"{user.email}: {user.first_name} {user.last_name}")

# List users with pagination
users = client.list_users(page=1, page_size=50)
 
print(f"Total users: {users.total}")
 
for user in users.users:
    print(f"{user.email}: {user.first_name} {user.last_name}")
    if user.metadata:
        print(f"  Metadata: {user.metadata}")

# Update user details
updated_user = client.update_user(
    "user@example.com",
    first_name="Jane",
    last_name="Smith"
)
 
print(f"Updated: {updated_user.first_name} {updated_user.last_name}")

# Soft delete user
client.delete_user("user@example.com")

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Create a user-scoped client
user_client = client.for_user("user@example.com")
 
# All operations now scoped to this user
response = user_client.invoke("What's the weather?")
 
# List user's threads
threads = user_client.list_threads()
 
# Upload file for user
file = user_client.upload_file(file_path="document.pdf")

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Use context manager for automatic cleanup
with client.as_user("user@example.com") as user_client:
    response = user_client.invoke("Analyze this data")
    print(response.output_text)

from lumnisai import AsyncClient
 
client = AsyncClient(api_key="your-api-key")
 
async with client.as_user("user@example.com") as user_client:
    response = await user_client.invoke("Hello!")
    print(response.output_text)

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Create users
alice = client.create_user(email="alice@example.com", first_name="Alice")
bob = client.create_user(email="bob@example.com", first_name="Bob")
 
# Alice starts a conversation
response1 = client.invoke(
    "What's the weather like?",
    user_id=alice.email
)
 
thread_id = response1.thread_id
 
# Bob continues the conversation
response2 = client.invoke(
    "What about tomorrow?",
    thread_id=thread_id,
    user_id=bob.email
)
 
# Get full conversation
thread = client.get_thread(thread_id)
print(f"Thread: {thread.title}")
 
# Get all responses in thread
responses = client.list_responses()
for resp in [r for r in responses.responses if r.thread_id == thread_id]:
    user = client.get_user(resp.user_id)
    print(f"{user.first_name}: {resp.output_text}")

from lumnisai import Client, AgentConfig
 
# Initialize client
client = Client(api_key="your-api-key")
 
# Create multiple users
users = [
    {"email": "alice@example.com", "first_name": "Alice", "last_name": "Johnson"},
    {"email": "bob@example.com", "first_name": "Bob", "last_name": "Smith"},
    {"email": "charlie@example.com", "first_name": "Charlie", "last_name": "Brown"}
]
 
for user_data in users:
    user = client.create_user(**user_data)
    print(f"Created: {user.email}")
 
# List all users
all_users = client.list_users(page_size=100)
print(f"\nTotal users: {all_users.total}")
 
# Perform operations for specific user
with client.as_user("alice@example.com") as alice_client:
    # Create response for Alice
    response = alice_client.invoke(
        "What are the latest AI trends?",
        agent_config=AgentConfig(
            coordinator_model_name="openai:gpt-4.1",
            use_cognitive_tools=True
        )
    )
    print(f"\nAlice's response: {response.output_text[:100]}...")
    
    # List Alice's threads
    threads = alice_client.list_threads()
    print(f"Alice has {len(threads.threads)} threads")
 
# Update user information
updated = client.update_user(
    "bob@example.com",
    first_name="Robert"
)
print(f"\nUpdated user: {updated.first_name} {updated.last_name}")
 
# Get user by ID
user = client.get_user(updated.id)
print(f"Retrieved by ID: {user.email}")

# Good: User-scoped operations
response = client.invoke(
    "Hello!",
    user_id="user@example.com"
)
 
# Or use scoped client
user_client = client.for_user("user@example.com")
response = user_client.invoke("Hello!")

from lumnisai import Client, NotFoundError, ValidationError
 
client = Client(api_key="your-api-key")
 
try:
    user = client.get_user("nonexistent@example.com")
except NotFoundError:
    # Create user if not found
    user = client.create_user(
        email="nonexistent@example.com",
        first_name="New",
        last_name="User"
    )
 
try:
    user = client.create_user(email="invalid-email")
except ValidationError as e:
    print(f"Validation error: {e.details}")

# Get all users with pagination
all_users = []
page = 1
page_size = 50
 
while True:
    result = client.list_users(page=page, page_size=page_size)
    all_users.extend(result.users)
    
    if len(result.users) < page_size:
        break
    
    page += 1
 
print(f"Total users retrieved: {len(all_users)}")