Integrations - Python SDK

from lumnisai import Client
 
client = Client(api_key="your-api-key")
 
# Initiate connection to GitHub
result = client.initiate_connection(
    user_id="user@example.com",
    app_name="GITHUB"
)
 
print(f"Visit this URL to connect: {result.redirect_url}")

# Get connection status
status = client.get_connection_status(
    user_id="user@example.com",
    app_name="GITHUB"
)
 
print(f"Status: {status.status}")
print(f"Connected at: {status.connected_at}")
 
if status.status == "active":
    print("Connection is active and ready to use")
elif status.status == "pending":
    print("Waiting for user to complete OAuth flow")
elif status.status == "failed":
    print(f"Connection failed: {status.error_message}")

# Wait for user to complete OAuth flow
try:
    status = client.wait_for_connection(
        user_id="user@example.com",
        app_name="GITHUB",
        timeout=300.0,  # 5 minutes
        poll_interval=5.0,
        target_status="active"
    )
    print(f"Connection established! Status: {status.status}")
except TimeoutError:
    print("User did not complete OAuth flow in time")

# Get all connections for a user
connections = client.list_connections(
    user_id="user@example.com"
)
 
print(f"Total connections: {len(connections.connections)}")
 
for conn in connections.connections:
    print(f"\n{conn.app_name}")
    print(f"  Status: {conn.status}")
    print(f"  Connected: {conn.connected_at}")
 
# Filter by specific apps
github_slack = client.list_connections(
    user_id="user@example.com",
    app_filter="GITHUB,SLACK"
)

# Get tools available based on user's connections
tools = client.get_integration_tools(
    user_id="user@example.com",
    app_filter=["GITHUB", "SLACK"]
)
 
print(f"Available tools: {tools.tool_count}")
 
for tool in tools.tools:
    print(f"\n{tool.name}")
    print(f"  Description: {tool.description}")
    print(f"  App: {tool.app_name}")

# List enabled apps for tenant
apps = client.list_apps(include_available=True)
 
print(f"Enabled apps: {', '.join(apps['enabled_apps'])}")
print(f"Total available: {apps['total_available']}")

# Check specific app
result = client.is_app_enabled("GITHUB")
 
if result['enabled']:
    print("GitHub integration is enabled for this tenant")
else:
    print("GitHub integration is not enabled")

# Enable an app for the tenant
client.set_app_enabled("LINEAR", enabled=True)
print("Linear integration enabled")
 
# Disable an app
client.set_app_enabled("LINEAR", enabled=False)
print("Linear integration disabled")

from lumnisai import Client
import time
 
client = Client(api_key="your-api-key")
 
# Step 1: Enable apps for tenant
apps_to_enable = ["GITHUB", "GMAIL", "GOOGLEDOCS", "GOOGLESHEETS"]
 
print("Enabling apps for tenant...")
for app in apps_to_enable:
    if not client.is_app_enabled(app)['enabled']:
        client.set_app_enabled(app, enabled=True)
        print(f"  ✓ Enabled {app}")
    else:
        print(f"  • {app} already enabled")
 
# Step 2: Initiate connections for user
user_email = "user@example.com"
 
print(f"\nInitiating connections for {user_email}...")
for app in apps_to_enable:
    # Check if already connected
    status = client.get_connection_status(user_email, app)
    
    if status.status != "active":
        # Initiate connection
        result = client.initiate_connection(
            user_id=user_email,
            app_name=app
        )
        print(f"  → {app}: {result.redirect_url}")
    else:
        print(f"  ✓ {app} already connected")
 
# Step 3: Wait for connections (in real app, user would click URLs)
print("\nWaiting for user to complete OAuth flows...")
for app in apps_to_enable:
    status = client.get_connection_status(user_email, app)
    if status.status == "active":
        print(f"  ✓ {app} connected")
 
# Step 4: List all connections
print("\nFinal connection status:")
connections = client.list_connections(user_email)
 
for conn in connections.connections:
    status_icon = "✓" if conn.status == "active" else "⨯"
    print(f"  {status_icon} {conn.app_name}: {conn.status}")
 
# Step 5: Get available tools
print("\nAvailable integration tools:")
tools = client.get_integration_tools(user_email)
 
for tool in tools.tools:
    print(f"  • {tool.name} ({tool.app_name})")

from lumnisai import Client, AgentConfig
 
client = Client(api_key="your-api-key")
 
# Ensure user has connected apps
user_email = "user@example.com"
tools = client.get_integration_tools(user_email)
 
print(f"User has {tools.tool_count} integration tools available")
 
# Create AI response with access to integration tools
response = client.invoke(
    """
    Search for recent pull requests in my GitHub repositories,
    then send a summary email via Gmail to team@example.com
    """,
    user_id=user_email,
    agent_config=AgentConfig(
        coordinator_model_name="openai:gpt-4.1",
        use_cognitive_tools=True
    )
)
 
print(response.output_text)

from lumnisai import Client
 
client = Client(api_key="your-api-key")
user_email = "user@example.com"
 
# Ensure Gmail and Google Calendar are connected
required_apps = ["GMAIL", "GOOGLECALENDAR"]
 
for app in required_apps:
    status = client.get_connection_status(user_email, app)
    if status.status != "active":
        print(f"Please connect {app}")
        result = client.initiate_connection(user_id=user_email, app_name=app)
        print(f"Visit: {result.redirect_url}")
 
# Use in AI task
from lumnisai import display_progress
 
for update in client.invoke(
    """
    Check my calendar for meetings today, then send me an email
    summary of all scheduled meetings with their times and participants.
    Send to: {user_email}
    """,
    stream=True,
    user_id=user_email
):
    display_progress(update)
    
    if update.state == "completed":
        print(f"\n{update.output_text}")

from lumnisai import AsyncClient
import asyncio
 
async def setup_integrations():
    client = AsyncClient(api_key="your-api-key")
    
    async with client:
        user_email = "user@example.com"
        apps = ["GITHUB", "SLACK", "GMAIL"]
        
        # Enable apps concurrently
        enable_tasks = [
            client.set_app_enabled(app, enabled=True)
            for app in apps
        ]
        await asyncio.gather(*enable_tasks)
        
        # Initiate connections concurrently
        connection_tasks = [
            client.initiate_connection(user_id=user_email, app_name=app)
            for app in apps
        ]
        results = await asyncio.gather(*connection_tasks)
        
        for app, result in zip(apps, results):
            print(f"{app}: {result.redirect_url}")
        
        # Wait for all connections
        wait_tasks = [
            client.wait_for_connection(
                user_id=user_email,
                app_name=app,
                timeout=300.0
            )
            for app in apps
        ]
        
        try:
            statuses = await asyncio.gather(*wait_tasks)
            print("\nAll connections established!")
        except TimeoutError as e:
            print(f"Some connections timed out: {e}")
 
asyncio.run(setup_integrations())

def ensure_connections(client, user_id: str, required_apps: list[str]):
    """Ensure user has required app connections."""
    missing = []
    
    for app in required_apps:
        status = client.get_connection_status(user_id, app)
        if status.status != "active":
            missing.append(app)
    
    if missing:
        print(f"Missing connections: {', '.join(missing)}")
        for app in missing:
            result = client.initiate_connection(user_id=user_id, app_name=app)
            print(f"Connect {app}: {result.redirect_url}")
        return False
    
    return True
 
# Usage
client = Client(api_key="your-api-key")
 
if ensure_connections(client, "user@example.com", ["GITHUB", "SLACK"]):
    # Proceed with task
    response = client.invoke(
        "Create a GitHub issue and notify team on Slack",
        user_id="user@example.com"
    )

from lumnisai import Client, LumnisAPIError
 
client = Client(api_key="your-api-key")
 
try:
    status = client.get_connection_status(
        user_id="user@example.com",
        app_name="GITHUB"
    )
    
    if status.status == "failed":
        print(f"Connection failed: {status.error_message}")
        # Re-initiate connection
        result = client.initiate_connection(
            user_id="user@example.com",
            app_name="GITHUB"
        )
        print(f"Please reconnect: {result.redirect_url}")
        
except LumnisAPIError as e:
    print(f"API error: {e}")

from datetime import datetime, timedelta
 
def check_connection_freshness(client, user_id: str, app_name: str, max_age_days: int = 30):
    """Check if connection is recent enough."""
    status = client.get_connection_status(user_id, app_name)
    
    if status.status != "active":
        return False
    
    if status.connected_at:
        connected_date = datetime.fromisoformat(status.connected_at.replace('Z', '+00:00'))
        age = datetime.now(connected_date.tzinfo) - connected_date
        
        if age > timedelta(days=max_age_days):
            print(f"{app_name} connection is {age.days} days old")
            return False
    
    return True
 
# Usage
if not check_connection_freshness(client, "user@example.com", "GITHUB", max_age_days=30):
    # Reconnect
    result = client.initiate_connection(user_id="user@example.com", app_name="GITHUB")
    print(f"Please reconnect: {result.redirect_url}")