layerscale 0.2.1

Python client for the LayerScale inference server

layerscale

Python client for the LayerScale inference server.

Install

pip install layerscale

Get a license key

You need a LayerScale license key to authenticate. Grab a free one at layerscale.ai/get-license — it takes about 10 seconds.

Usage

from layerscale import LayerScale

client = LayerScale(
    "http://localhost:8080",
    api_key="LS-...",  # or set LAYERSCALE_LICENSE_KEY env var
)

Complete example

End-to-end: create a session, push a few OHLCV candles, wait for them to be processed, and ask the model about the data.

import os
from layerscale import LayerScale

client = LayerScale(
    "http://localhost:8080",
    api_key=os.environ.get("LAYERSCALE_LICENSE_KEY"),
)

# 1. Create a session with a system prompt and freeze it in the cache
#    so it is not reprocessed on every incoming data update.
session = client.sessions.create(
    type="ohlcv",
    prompt=(
        "You are a real-time market analyst. You receive live OHLCV candles "
        "and answer questions about market direction in a single word when possible."
    ),
    context=4096,
    mark_prefix=True,
)

# 2. Push a few OHLCV candles and wait for them to be decoded into the cache.
#    For a reactive alternative, iterate `client.sessions.events()` or open a
#    WebSocket via `client.sessions.stream()` instead of using `wait=True`.
candles = [
    {"o": 150.20, "h": 150.80, "l": 150.10, "c": 150.70, "v": 1_200_000, "timestamp": 1_733_000_000, "sym": "AAPL"},
    {"o": 150.70, "h": 151.10, "l": 150.60, "c": 150.95, "v":   900_000, "timestamp": 1_733_000_060, "sym": "AAPL"},
    {"o": 150.95, "h": 151.40, "l": 150.90, "c": 151.30, "v": 1_500_000, "timestamp": 1_733_000_120, "sym": "AAPL"},
    {"o": 151.30, "h": 151.80, "l": 151.20, "c": 151.75, "v": 1_100_000, "timestamp": 1_733_000_180, "sym": "AAPL"},
    {"o": 151.75, "h": 152.10, "l": 151.60, "c": 152.00, "v": 1_300_000, "timestamp": 1_733_000_240, "sym": "AAPL"},
]
client.sessions.push(session.session_id, candles, wait=True)

# 3. Query the model about the data we just ingested.
answer = client.sessions.query(
    session.session_id,
    prompt="Is the market trending up or down?",
    max_tokens=8,
    fast_answer=True,
)
print("Answer:", answer.text.strip())

client.sessions.delete(session.session_id)

Async client

import asyncio
from layerscale import AsyncLayerScale

async def main():
    async with AsyncLayerScale("http://localhost:8080") as client:
        session = await client.sessions.create(type="ohlcv", prompt="...", mark_prefix=True)
        await client.sessions.push(session.session_id, candles, wait=True)
        answer = await client.sessions.query(session.session_id, prompt="Trend?")
        print(answer.text)
        await client.sessions.delete(session.session_id)

asyncio.run(main())

OpenAI-compatible chat

chat = client.chat.create(
    messages=[{"role": "user", "content": "Hello"}],
)
print(chat.choices[0].message.content)

Anthropic-compatible messages

msg = client.messages.create(
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=256,
)
for block in msg.content:
    if block.type == "text":
        print(block.text)

Streaming

# Chat completions
with client.chat.stream(messages=[{"role": "user", "content": "Tell me a joke"}]) as stream:
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)

# Anthropic-style messages
with client.messages.stream(
    messages=[{"role": "user", "content": "Tell me a joke"}],
    max_tokens=256,
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)

Sessions

session = client.sessions.create(
    type="ohlcv",
    prompt="You are a trading analyst.",
    flash=[{"query": "Is the trend bullish?"}, {"query": "Is volume increasing?"}],
    mark_prefix=True,
)

answer = client.sessions.query(
    session.session_id,
    prompt="What is the trend?",
    max_tokens=256,
)

# Stream generation token-by-token
with client.sessions.query_stream(session.session_id, max_tokens=256) as stream:
    for chunk in stream:
        if not chunk.done:
            print(chunk.token, end="", flush=True)

Tool calling in sessions

Pass OpenAI-format messages + tools and the server applies the model's Jinja chat template (so tool-result headers render exactly like they do on /v1/chat/completions). The server also diffs against the session's cached tokens and only decodes the delta — turn N is as cheap as turn 1 + a token compare. The tool_call_guide is cached per session so repeated turns with the same tool set don't re-tokenize tool names.

tools = [{
    "type": "function",
    "function": {
        "name": "read_file",
        "description": "Read a file",
        "parameters": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
        },
    },
}]

# Turn 1: ask the model to pick a tool
resp = client.sessions.query(
    session.session_id,
    messages=[
        {"role": "system", "content": "You are a coding agent."},
        {"role": "user", "content": "Read src/App.tsx"},
    ],
    tools=tools,
    max_tokens=256,
)

if resp.tool_calls:
    call = resp.tool_calls[0]
    print(f"Tool: {call.function.name}({call.function.arguments})")

    # Turn 2: feed the tool result back. Include the prior assistant call
    # so the template can render the matching tool_call_id header.
    follow_up = client.sessions.query(
        session.session_id,
        messages=[
            {"role": "system", "content": "You are a coding agent."},
            {"role": "user", "content": "Read src/App.tsx"},
            {"role": "assistant", "tool_calls": [call.model_dump()]},
            {"role": "tool", "tool_call_id": call.id,
             "content": "import React from 'react'; ..."},
        ],
        tools=tools,
        max_tokens=256,
    )
    print(follow_up.text)

Session management

sessions = client.sessions.list()
state = client.sessions.get(session.session_id)
client.sessions.delete(session.session_id)

Continuous streaming

Push OHLCV data into the lock-free ring buffer for background processing:

client.sessions.push(session.session_id, [
    {"o": 150.5, "h": 151.0, "l": 150.0, "c": 150.75, "v": 1_000_000, "timestamp": 1704067200, "sym": "AAPL"},
])

status = client.sessions.stream_status(session.session_id)
stats  = client.sessions.stats(session.session_id)

Flash queries

Register questions that are automatically evaluated in the background after each data update. Answers are cached for near-instant retrieval:

registered = client.sessions.flash(session.session_id, "Is the trend bullish?")
# optional `max_tokens` caps the answer length (default 32):
client.sessions.flash(session.session_id, "Summarise the trend", max_tokens=64)

queries = client.sessions.list_flash(session.session_id)
client.sessions.unflash(session.session_id, registered.id)

Session events (SSE)

with client.sessions.events(session.session_id) as events:
    for event in events:
        if event.type == "flash_ready":
            print(event.query, event.value, event.confidence)
        elif event.type == "data_updated":
            print("new data version:", event.data_version)

WebSocket streaming

A WebSocket combines data-push and event subscription on one connection. It does not auto-reconnect — wrap it in your own backoff loop for long-lived consumers:

from layerscale import WsFlashReady, WsDataUpdated

with client.sessions.stream(session.session_id) as socket:
    socket.push([candle])
    for event in socket:
        if isinstance(event, WsFlashReady):
            print(event.data.query, event.data.value)
        elif isinstance(event, WsDataUpdated):
            print("version:", event.data.data_version)

Cancellation and timeouts

The default request timeout is 10 minutes. Override it on the client or per-request:

client = LayerScale("http://localhost:8080", timeout=30.0)  # 30s default

# Per-request override
answer = client.sessions.query(session_id, prompt="...")  # uses client default

Error handling

from layerscale import APIStatusError, LayerScaleError

try:
    client.sessions.query(session_id, max_tokens=256)
except APIStatusError as exc:
    print(exc.status_code, exc.message)
except LayerScaleError as exc:
    print(exc)

API

Method	Endpoint	Description
health.check()	GET /v1/health	Check whether the server is ready.
models.list()	GET /v1/models	List loaded models.
chat.create(**params)	POST /v1/chat/completions	OpenAI-compatible chat completion.
chat.stream(**params)	POST /v1/chat/completions (SSE)	Stream chat chunks.
completions.create(**params)	POST /v1/completions	OpenAI-compatible legacy completion.
messages.create(**params)	POST /v1/messages	Anthropic-compatible messages API.
messages.stream(**params)	POST /v1/messages (SSE)	Stream messages events.
sessions.create(**params)	POST /v1/sessions/init	Create a streaming session.
sessions.list()	GET /v1/sessions	List all active sessions.
sessions.get(id)	GET /v1/sessions/:id/state	Inspect a session's state.
sessions.delete(id)	DELETE /v1/sessions/:id	Delete a session.
sessions.append(id, text)	POST /v1/sessions/:id/append	Append raw text to context.
sessions.slots(id)	GET /v1/sessions/:id/slots	List cache slots.
sessions.query(id, **params)	POST /v1/sessions/:id/generate	Run a generation.
sessions.query_stream(id, **params)	POST /v1/sessions/:id/generate (SSE)	Stream generation token-by-token.
sessions.mark_prefix(id)	POST /v1/sessions/:id/mark_prefix	Freeze cache prefix.
sessions.push(id, data, wait=?)	POST /v1/sessions/:id/stream/push	Push streaming data.
sessions.stream_status(id)	GET /v1/sessions/:id/stream/status	Processor status.
sessions.stats(id)	GET /v1/sessions/:id/stats	Computed statistics.
sessions.flash(id, query, max_tokens=?)	POST /v1/sessions/:id/flash	Register a flash query.
sessions.list_flash(id)	GET /v1/sessions/:id/flash	List flash queries.
sessions.unflash(id, query_id)	DELETE /v1/sessions/:id/flash/:query_id	Remove a flash query.
sessions.events(id)	GET /v1/sessions/:id/events (SSE)	Subscribe to session events.
sessions.stream(id)	WS /v1/sessions/:id/ws	Bidirectional WebSocket.