omnia-sdk

Python client for the Omnia API.

Installation

pip install omnia-sdk

Quick start

from omnia_sdk import OmniaClient

client = OmniaClient(api_key="your-api-key")

me = client.auth.get_me()
project, chat = client.projects.create_project_with_chat(me.id, "My project")

for event in client.messages.stream_message(me.id, project.id, chat.id, "Hello"):
    print(event)

Configuration

The client can be configured via constructor arguments or environment variables:

Set OMNIA_ENV=dev to point automatically to http://localhost:8000, or OMNIA_ENV=staging for https://api.staging.omniasec.ai.

# All options explicit
client = OmniaClient(
    api_key="your-key",
    api_url="https://api.omniasec.ai",
    default_model="gemini-flash-latest",
    default_provider="google",
)

Client reference

client.auth

me = client.auth.get_me()                                  # -> User
agents = client.auth.get_agents()                          # -> list[dict]
settings = client.auth.get_last_user_settings(user_id)     # -> dict
settings = client.auth.update_user_settings(user_id, {})   # -> dict

client.projects

projects = client.projects.list_projects(user_id)                        # -> list[Project]
project = client.projects.create_project(user_id, "Name")                # -> Project
project = client.projects.get_project(user_id, project_id)               # -> Project
client.projects.delete_project(user_id, project_id)

chats = client.projects.list_chats(user_id, project_id)                  # -> list[Chat]
chat = client.projects.create_chat(user_id, project_id)                  # -> Chat
chat = client.projects.get_or_create_chat(user_id, project_id)           # -> Chat
project, chat = client.projects.create_project_with_chat(user_id, "Name")

client.messages

messages = client.messages.list_messages(user_id, project_id, chat_id)   # -> list[ChatMessage]
client.messages.delete_message(user_id, project_id, chat_id, message_id)

# Streaming (SSE) — yields raw event dicts
for event in client.messages.stream_message(user_id, project_id, chat_id, "query"):
    print(event)

# Non-streaming
response = client.messages.send_message(user_id, project_id, chat_id, "query")

# Launch a workflow from a message
client.messages.launch_workflow(
    agent_name="OmniaMainAgent",
    workflow_endpoint="/workflow/run",
    user_id=user_id,
    project_id=project_id,
    chat_id=chat_id,
    workflow_template_id="template-id",
    written_params={"key": "value"},
)

client.resources

from pathlib import Path

resources = client.resources.list_resources(user_id, project_id)             # -> list[Resource]
resource = client.resources.upload_resource(user_id, project_id, Path("file.pdf"))  # -> Resource
client.resources.delete_resource(user_id, project_id, resource_id)

client.analysis

analyses = client.analysis.list_analyses(user_id)                # -> list[Analysis]
analyses = client.analysis.list_analyses(user_id, shared=True)   # -> list[Analysis]
analysis = client.analysis.get_analysis(analysis_id)             # -> Analysis
children = client.analysis.get_analysis_children(analysis_id)    # -> list[Analysis]
analysis = client.analysis.upload_file(user_id, Path("sample.exe"))  # -> Analysis

client.templates

templates = client.templates.list_templates(user_id)                    # -> list[Template]
template = client.templates.get_template(user_id, template_id)          # -> Template
template = client.templates.create_template(user_id, payload)           # -> Template
template = client.templates.update_template(user_id, template_id, payload)  # -> Template
client.templates.delete_template(user_id, template_id)
template = client.templates.fork_template(user_id, template_id)         # -> Template

client.workflows

result = client.workflows.run_file_analysis(user_id, project_id, resource_hash, chat_id)

client.public

No authentication required for these endpoints.

version = client.public.get_api_version()
analysis = client.public.get_public_analysis(analysis_id)
children = client.public.get_public_analysis_children(analysis_id)
template = client.public.get_public_template(template_id)
results = client.public.search_market("react")
package = client.public.get_market_package("npm", "lodash")
versions = client.public.get_market_package_versions("npm", "lodash")
chat = client.public.get_shared_chat(token)

Error handling

from omnia_sdk import OmniaAPIError, NotConfiguredError

try:
    me = client.auth.get_me()
except NotConfiguredError:
    print("No API key configured")
except OmniaAPIError as e:
    print(f"API error {e.status_code}: {e.detail}")

Version

import omnia_sdk
print(omnia_sdk.__version__)

License

MIT