Python SDK

pip install omnara

from omnara import OmnaraClient
import uuid

# Initialize client
client = OmnaraClient(api_key="your-api-key")

# Generate session ID
instance_id = str(uuid.uuid4())

# Send a status update
response = client.send_message(
    agent_type="My Custom Agent",
    content="Analyzing codebase structure",
    agent_instance_id=instance_id,
    requires_user_input=False
)

print(f"Message sent: {response['message_id']}")

# Ask for user input
answer = client.send_message(
    content="Should I refactor this legacy module?",
    agent_instance_id=instance_id,
    requires_user_input=True
)

print(f"User answered: {answer}")

from omnara import OmnaraClient

# Basic
client = OmnaraClient(api_key="xxxxxxxxxxxxxxxxxxxx")

# Self-hosted
client = OmnaraClient(
    api_key="xxxxxxxxxxxxxxxxxxxx",
    base_url="https://omnara.mycompany.com"
)

# From environment variable
import os
client = OmnaraClient(api_key=os.getenv("OMNARA_API_KEY"))

response = client.send_message(
    agent_type="Code Analyzer",
    content="Found 5 potential issues",
    agent_instance_id=instance_id,
    requires_user_input=False,
    send_push=True  # Send push notification
)

# Check for queued user messages
if response.get("queued_user_messages"):
    for msg in response["queued_user_messages"]:
        print(f"User said: {msg['content']}")

answer = client.send_message(
    content="Which file should I start with?",
    agent_instance_id=instance_id,
    requires_user_input=True,
    timeout_minutes=60,  # Wait up to 1 hour
    poll_interval=10.0   # Check every 10 seconds
)

print(f"User selected: {answer}")

from omnara import AsyncOmnaraClient
import asyncio

async def main():
    client = AsyncOmnaraClient(api_key="your-api-key")

    # Send message
    response = await client.send_message(
        agent_type="Async Agent",
        content="Processing...",
        requires_user_input=False
    )

    # Ask question
    answer = await client.send_message(
        content="Proceed?",
        agent_instance_id=response["agent_instance_id"],
        requires_user_input=True
    )

    print(answer)

asyncio.run(main())

import uuid

# Generate unique session ID
session_id = str(uuid.uuid4())

# First message creates session
client.send_message(
    agent_type="My Agent",
    content="Starting work",
    agent_instance_id=session_id,
    requires_user_input=False
)

client.end_session(agent_instance_id=session_id)

import subprocess

def get_git_diff():
    result = subprocess.run(
        ["git", "diff", "HEAD"],
        capture_output=True,
        text=True
    )
    return result.stdout

# Send with diff
response = client.send_message(
    agent_type="Code Agent",
    content="Modified authentication flow",
    agent_instance_id=instance_id,
    requires_user_input=False,
    git_diff=get_git_diff()
)

response = client.send_message(
    agent_type="Test Runner",
    content="Tests completed",
    agent_instance_id=instance_id,
    requires_user_input=False,
    message_metadata={
        "test_results": {
            "passed": 42,
            "failed": 3,
            "skipped": 1
        },
        "duration_ms": 1234,
        "commit_hash": "abc123def"
    }
)

from omnara import OmnaraClient
from omnara.exceptions import OmnaraAPIError

client = OmnaraClient(api_key="your-key")

try:
    response = client.send_message(
        agent_type="My Agent",
        content="Test message",
        requires_user_input=False
    )
except OmnaraAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")
    # Handle error appropriately
except Exception as e:
    print(f"Unexpected error: {e}")

from omnara import OmnaraClient
import uuid
import time

def run_agent():
    # Initialize
    client = OmnaraClient(api_key="your-key")
    session_id = str(uuid.uuid4())

    try:
        # Start session
        client.send_message(
            agent_type="Code Review Agent",
            content="Starting code review",
            agent_instance_id=session_id,
            requires_user_input=False
        )

        # Do some work
        time.sleep(2)

        # Report progress
        client.send_message(
            content="Analyzed 10 files, found 3 issues",
            agent_instance_id=session_id,
            requires_user_input=False
        )

        # Ask for guidance
        answer = client.send_message(
            content="Should I auto-fix the linting issues?",
            agent_instance_id=session_id,
            requires_user_input=True,
            timeout_minutes=30
        )

        # Process response
        if "yes" in answer.lower():
            client.send_message(
                content="Fixing issues...",
                agent_instance_id=session_id,
                requires_user_input=False
            )
            # Fix issues...
            time.sleep(5)
            client.send_message(
                content="All issues fixed! ✅",
                agent_instance_id=session_id,
                requires_user_input=False,
                send_push=True
            )
        else:
            client.send_message(
                content="Skipping auto-fix as requested",
                agent_instance_id=session_id,
                requires_user_input=False
            )

    finally:
        # Always end session
        client.end_session(agent_instance_id=session_id)

if __name__ == "__main__":
    run_agent()

from omnara import AsyncOmnaraClient

async def async_agent():
    client = AsyncOmnaraClient(api_key="your-key")

    response = await client.send_message(
        agent_type="Async Agent",
        content="Processing async task",
        requires_user_input=False
    )

    answer = await client.send_message(
        content="Continue processing?",
        agent_instance_id=response["agent_instance_id"],
        requires_user_input=True
    )

    await client.end_session(agent_instance_id=response["agent_instance_id"])