orchestrator-client

Python client for the DAMIT AIOps Orchestrator REST API and Socket.IO realtime events.

Install

pip install orchestrator-client

Requires Python >= 3.12.

Quick Start

from orchestrator_client import Orchestrator

client = Orchestrator(base_url="http://localhost:8080")

# Create a task
task = client.create_task(
    workflow_id="proactive",
    goal_prompt="Analyze system logs for errors",
    max_iterations=50,
)
print(f"Created: {task.task_id}")

# Poll status
status = client.get_task_status(task.task_id)
print(f"Status: {status.status}, iteration {status.iteration}/{status.max_iterations}")

# List tasks
tasks = client.list_tasks(workflow_id="proactive", limit=10)
for t in tasks.tasks:
    print(f"  {t.id}: {t.status}")

# Cancel
client.cancel_task(task.task_id)

client.close()

Authentication

The SDK sends a bearer token on every request when configured; the Socket.IO client carries it in the handshake (auth.token) so the server can authenticate the connection under socket_auth_policy: strict:

from orchestrator_client import Orchestrator
from orchestrator_client.socketio import RealtimeClient

# REST: static token on every request
client = Orchestrator(base_url="http://localhost:8080", api_key="<keycloak-token>")

# Realtime: static token, or a callable (re-invoked on every connect/reconnect)
rt = RealtimeClient("http://localhost:8080", get_token=lambda: get_token_from_memory())
await rt.connect()

When neither is set, no Authorization header / auth payload is sent — the SDK behaves exactly as before.

Dynamic service tokens (ServiceTokenSource)

For gateway-relayed integration apps, the Keycloak JWT is minted via the client-credentials grant and sent as X-Authorization on REST requests alongside the gateway key in Authorization (the gateway strips only Authorization). The static api_key is not used for authentication in this mode:

from orchestrator_client import OrchestratorAsync
from orchestrator_client import ServiceTokenSource

source = ServiceTokenSource(
    base_url="http://localhost:8080",   # optional if KEYCLOAK_URL/KEYCLOAK_REALM are set
    client_id="ai-wiki",
    client_secret="<service-secret>",
)
client = OrchestratorAsync(base_url="http://localhost:8080", x_token_getter=source.get_token)

The getter is resolved per request and re-invoked on every connect/reconnect, so an expired JWT refreshes itself. ServiceTokenSource caches the mint with a freshness margin and a negative cache, and raises KeycloakTokenError on failure (fetch rejection, non-OK response, malformed payload, negative-cache hit). Close the source on shutdown (await source.aclose()) when it owns its HTTP client.

RealtimeClient

The RealtimeClient carries credentials in the Socket.IO handshake auth payload:

• get_token — a static token (auth.token) or a callable re-invoked on every connect/reconnect (async callables are awaited);
• api_key — the gateway API key, sent as Authorization: Bearer <key> in the auth payload next to token, for gateway deployments that validate the handshake against the gateway key alongside the JWT.

from orchestrator_client.socketio import RealtimeClient

rt = RealtimeClient(
    "http://localhost:8080",
    get_token=lambda: get_token_from_memory(),
    api_key="<gateway-key>",
)
await rt.connect()

Token-getter failures are contained: a throwing getter degrades to an unauthenticated handshake, which the server rejects — the client never hangs on the connect.

Environment Variables

SSL / Self-Signed Certificates

For orchestrator instances behind HTTPS with self-signed certificates, pass verify_ssl=False:

from orchestrator_client import Orchestrator

client = Orchestrator(
    base_url="https://orchestrator.internal:8443",
    verify_ssl=False,
)

Or inject a pre-configured httpx.AsyncClient for full control:

import httpx
from orchestrator_client import OrchestratorAsync

custom = httpx.AsyncClient(verify=False, timeout=httpx.Timeout(60.0))
client = OrchestratorAsync(http_client=custom)

## Async Variant

For use inside async code (e.g. FastAPI, asyncio scripts):

```python
from orchestrator_client import OrchestratorAsync

async with OrchestratorAsync() as client:
    status = await client.get_task_status("task-abc123")

Exceptions

All inherit from OrchestratorError and carry status_code and error_code:

Documentation

Detailed docs with full method listings and examples:

• REST API client — all endpoints, workflow interactions, configuration
• Socket.IO realtime — event types, room subscriptions, streaming
• SSE status stream — lightweight status-only monitoring

Testing

pip install -e ".[dev]"
pytest

License

Apache 2.0