Python SDK

Official Python SDK for Scalix Cloud

Install

bash

pip install scalix-sdk

To update to the latest release: pip install --upgrade scalix-sdk. Versions follow the platform API line; minor and patch updates are backwards-compatible.

The SDK is generated from the platform's OpenAPI spec (API v1.3.4 — 237 operations across 31 service areas), so every gateway endpoint is available as a typed function.

Initialize

Create one AuthenticatedClient and pass it to every operation:

python

import os
from scalix_sdk import AuthenticatedClient
 
client = AuthenticatedClient(
    base_url="https://api.scalix.world",
    token=os.environ["SCALIX_API_KEY"],
)

Operations are grouped by service under scalix_sdk.generated.api.<tag> (account, database, functions, storage, ai, vectors, key_value, …). Each operation module exposes four callables:

Callable	Behavior
`sync(client=...)`	Blocking; returns the parsed response model (or `None`/`ErrorResponse`)
`sync_detailed(client=...)`	Blocking; returns a `Response` with `status_code`, `content`, `headers`, `parsed`
`asyncio(client=...)`	`await`-able version of `sync`
`asyncio_detailed(client=...)`	`await`-able version of `sync_detailed`

For endpoints whose success body isn't typed in the spec (functions, storage, KV, vectors), sync() returns no parsed model — use sync_detailed() and decode resp.content yourself, as shown below.

Account

GET /v1/me describes the current token — identity, effective permissions, the MCP tools your scopes unlock, and plan limits:

python

from scalix_sdk.generated.api.account import get_me
from scalix_sdk.generated.models import MeResponse
 
me = get_me.sync(client=client)
if isinstance(me, MeResponse):
    print(me.identity.org_id, me.identity.project_id)
    print(me.tools)  # MCP tools available to this token

Database

Run SQL against your database with execute_sql. The response is fully typed:

python

from scalix_sdk.generated.api.database import execute_sql
from scalix_sdk.generated.models import SqlRequest, SqlResponse
 
result = execute_sql.sync(
    client=client,
    body=SqlRequest(query="SELECT id, email FROM users WHERE active = true"),
)
if isinstance(result, SqlResponse):
    print(result.columns)      # ["id", "email"]
    print(result.row_count)    # number of rows returned
    print(result.duration_ms)  # server-side execution time

SqlRequest also takes a params field for $1, $2, … placeholders, and batch_sql executes multiple statements in one round trip. The database tag also includes natural_language_to_sql, optimize_query, and detect_anomalies.

Functions

List your functions and invoke one by id or name:

python

import json
from scalix_sdk.generated.api.functions import invoke_function, list_functions
from scalix_sdk.generated.models import InvokeFunctionBody
 
resp = list_functions.sync_detailed(client=client)
functions = json.loads(resp.content)
 
body = InvokeFunctionBody.from_dict({"order_id": "ord_123"})
resp = invoke_function.sync_detailed("process-order", client=client, body=body)
result = json.loads(resp.content)

The functions tag also covers create_function, update_function, get_function_logs, and list_function_invocations.

Storage

Objects are raw bytes — wrap uploads in the generated File type:

python

import json
from io import BytesIO
from scalix_sdk.generated.api.storage import get_object, list_buckets, put_object
from scalix_sdk.generated.types import File
 
put_object.sync_detailed(
    "my-bucket", "notes/hello.txt",
    client=client,
    body=File(payload=BytesIO(b"hello world")),
)
 
resp = get_object.sync_detailed("my-bucket", "notes/hello.txt", client=client)
data = resp.content  # raw object bytes
 
buckets = json.loads(list_buckets.sync_detailed(client=client).content)

create_bucket, delete_object, head_object, and presign_object (presigned URLs) are also available under the storage tag.

AI Chat Completions

The request/response envelope is OpenAI-compatible:

python

from scalix_sdk.generated.api.ai import chat_completion
from scalix_sdk.generated.models import ChatCompletionRequest, ChatCompletionResponse, ChatMessage
 
response = chat_completion.sync(
    client=client,
    body=ChatCompletionRequest(
        model="scalix-lumio-lite",
        messages=[ChatMessage(role="user", content="Hello!")],
    ),
)
if isinstance(response, ChatCompletionResponse):
    print(response.model)
    print(response.choices.to_dict())  # completion choices (message + finish reason)
    print(response.usage.to_dict())    # prompt/completion/total token counts

For streaming, stream_chat_completion posts to /v1/ai/chat/completions/stream and returns the raw server-sent-events body via sync_detailed(...).content. The ai tag also includes create_embeddings, list_models, image generation, speech, transcription, RAG, research, and text utilities.

Vectors

The vectors surface supports collections, single and batch upsert (one multi-row write), and nearest-neighbour search with metadata filtering. Operations are scoped to your database tenant id (UUID):

python

import json
from scalix_sdk.generated.api.vectors import batch_upsert_vectors, search_vectors
from scalix_sdk.generated.models import (
    VectorEntry,
    VectorEntryMetadataType0,
    VectorSearchRequest,
    VectorSearchRequestFilterType0,
    VectorUpsertRequest,
)
 
tenant_id = "00000000-0000-0000-0000-000000000000"  # your database tenant UUID
 
# Batch upsert with per-vector metadata
batch_upsert_vectors.sync_detailed(
    tenant_id,
    client=client,
    body=VectorUpsertRequest(
        collection="docs",
        entries=[
            VectorEntry(
                id="doc-1",
                embedding=[0.12, 0.51, 0.33],  # your embedding values
                metadata=VectorEntryMetadataType0.from_dict({"lang": "en"}),
            ),
        ],
    ),
)
 
# Search with a metadata filter
resp = search_vectors.sync_detailed(
    tenant_id,
    client=client,
    body=VectorSearchRequest(
        collection="docs",
        query_vector=[0.10, 0.48, 0.30],
        top_k=5,
        filter_=VectorSearchRequestFilterType0.from_dict({"lang": "en"}),
    ),
)
matches = json.loads(resp.content)  # search results with scores

Generate embeddings with create_embeddings under the ai tag, then upsert them here. upsert_vectors and list_vector_collections round out the surface.

KV Store

Values are JSON objects; ttl is optional, in seconds:

python

import json
from scalix_sdk.generated.api.key_value import kv_get, kv_set
from scalix_sdk.generated.models import KvKeyRequest, KvSetRequest, KvSetRequestValue
 
kv_set.sync_detailed(
    client=client,
    body=KvSetRequest(
        key="user:123",
        value=KvSetRequestValue.from_dict({"name": "Alice"}),
        ttl=3600,
    ),
)
 
resp = kv_get.sync_detailed(client=client, body=KvKeyRequest(key="user:123"))
value = json.loads(resp.content)

kv_del (takes a keys list), kv_exists, and kv_keys are also available.

Async Usage

Every operation has asyncio / asyncio_detailed twins that share the same client:

python

import asyncio
import os
from scalix_sdk import AuthenticatedClient
from scalix_sdk.generated.api.account import get_me
from scalix_sdk.generated.api.database import execute_sql
from scalix_sdk.generated.models import SqlRequest
 
async def main():
    client = AuthenticatedClient(
        base_url="https://api.scalix.world",
        token=os.environ["SCALIX_API_KEY"],
    )
    me = await get_me.asyncio(client=client)
    result = await execute_sql.asyncio(
        client=client,
        body=SqlRequest(query="SELECT 1"),
    )
 
asyncio.run(main())

Error Handling

Every endpoint returns the standard error envelope on failure — documented error statuses parse into ErrorResponse:

python

from scalix_sdk.generated.api.account import get_me
from scalix_sdk.generated.models import ErrorResponse
 
me = get_me.sync(client=client)
if isinstance(me, ErrorResponse):
    print(me.code)        # stable machine-readable code, e.g. "AUTH_FAILED"
    print(me.error)       # human-readable message
    print(me.request_id)  # correlation id for support

Use the _detailed variants when you need the status code and headers:

python

resp = get_me.sync_detailed(client=client)
print(resp.status_code)  # resp.content and resp.headers are also available

To raise on any status the spec doesn't document, construct the client with raise_on_unexpected_status=True and catch UnexpectedStatus:

python

from scalix_sdk import AuthenticatedClient
from scalix_sdk.generated import errors
 
client = AuthenticatedClient(
    base_url="https://api.scalix.world",
    token=os.environ["SCALIX_API_KEY"],
    raise_on_unexpected_status=True,
)
 
try:
    me = get_me.sync(client=client)
except errors.UnexpectedStatus as e:
    print(e.status_code, e.content)