thalovant 0.4.9

Python SDK and CLI for direct Thalovant hub data-plane clients and agents

Thalovant Python SDK

Python SDK for connecting apps, services, kiosks, and agents to Thalovant hubs.

The control API is used to discover hubs and provision a client identity. After that, the SDK talks directly to the hub data plane over HTTPS, WSS, or MQTTS.

Thalovant API      -> discover hubs, create client identity
Python SDK         -> connect to the hub data plane
Hub runtime        -> skills, events, replies

Full docs: https://docs.thalovant.com/developers/sdks/python/

What You Need

• A Thalovant account with API access for authenticated control-plane actions.
• A hub id or slug.
• A client identity for that hub. You can create one through the API or use one downloaded from the dashboard.

Install

pip install thalovant

For local SDK development:

pip install -e ".[dev]"

Quick Start

This is the normal first integration flow.

from thalovant import ThalovantClient, ThalovantControlPlane

api = ThalovantControlPlane("https://dash.thalovant.com/api")

# Public hub discovery does not require auth.
public_hubs = api.list_public_hubs(limit=12)
for hub in public_hubs["data"]:
    print(hub["id"], hub["slug"], hub["title"])

# Auth is required when creating a client identity.
api.login("[e-mail]", "password")

result = api.create_client_identity(
    "hub-id",
    name="python-demo-client",
    preferred_protocols=("wss", "https", "mqtt"),
)

with ThalovantClient(result.identity, protocol="wss") as client:
    reply = client.ask("Tell me a short clean joke.")
    print(reply.text)

Keep result.identity secret. It contains the client credentials used by the hub. Do not log result.identity.as_dict(include_secrets=True).

List Your Hubs

Authenticated accounts can list owned or visible hubs:

api = ThalovantControlPlane("https://dash.thalovant.com/api")
api.login("[e-mail]", "password")

page = api.list_hubs(limit=50)
for hub in page["data"]:
    print(hub["id"], hub["slug"], hub["title"])

Use An Existing Identity

If you already downloaded an identity from the dashboard or stored one from a previous provisioning step:

from thalovant import ThalovantClient

with ThalovantClient.from_identity_file("_identity.json") as client:
    reply = client.ask("What can this hub do?")
    print(reply.text)

Environment variables are supported too:

export THALOVANT_ACCESS_KEY=...
export THALOVANT_PASSWORD=...
export THALOVANT_CRYPTO_KEY=...
export THALOVANT_SITE_ID=...
export THALOVANT_HUB_HTTPS_HOST=https://hub.example.com
export THALOVANT_HUB_WSS_HOST=wss://hub.example.com
export THALOVANT_HUB_MQTT_HOST=mqtts://mqtt.thalovant.com:8883
export THALOVANT_MQTT_USERNAME=...
export THALOVANT_MQTT_PASSWORD=...
export THALOVANT_MQTT_TOPIC_PREFIX=hivemind/hub-id/client-id

from thalovant import ThalovantClient

with ThalovantClient.from_env(protocol="https") as client:
    print(client.ask("Say hello.").text)

Save A Provisioned Identity

Only save identities in a secret store or local developer file that is ignored by git.

import json
from pathlib import Path

Path("_identity.json").write_text(
    json.dumps(result.identity.as_dict(include_secrets=True), indent=2),
    encoding="utf-8",
)

Protocols

Hubs may expose one or more public data-plane protocols:

• wss: secure realtime WebSocket, the default public path.
• https: request/response HTTP protocol exposed as HTTPS.
• mqtt: broker-mediated MQTT over TLS. Requires per-client broker credentials.

Inspect what an identity supports:

identity = result.identity

print(identity.enabled_protocols())
print(identity.endpoint_for("wss"))
print(identity.endpoint_for("https"))
print(identity.endpoint_for("mqtt"))
print(identity.mqtt.endpoint if identity.mqtt else None)

Connect with a specific protocol:

from thalovant import ThalovantClient

for protocol in ("wss", "https", "mqtt"):
    if not result.identity.supports_protocol(protocol):
        continue
    if protocol == "mqtt" and result.identity.mqtt is None:
        continue

    with ThalovantClient(result.identity, protocol=protocol) as client:
        print(protocol, client.ask(f"Reply over {protocol}.").text)

MQTT identities include a broker endpoint, username, password, TLS flag, and topic prefix. The broker credentials are scoped to that client and should be treated like a password.

Conversations

Use a conversation when several turns should share one session.

from thalovant import ThalovantClient

with ThalovantClient.from_identity_file("_identity.json") as client:
    with client.conversation(lang="en-us") as convo:
        print(convo.ask("Remember that my favorite color is blue.").text)
        print(convo.ask("What color did I mention?").text)

Events

You can wait for, stream, or subscribe to hub events.

from thalovant import EVENT_SPEAK, ThalovantClient

with ThalovantClient.from_identity_file("_identity.json") as client:
    for event in client.listen(EVENT_SPEAK, timeout=30, max_events=1):
        print(event.text)

Use timeouts in scripts so they do not wait forever.

Client Context

Context lets skills know which app, device, user, or channel made the request.

from thalovant import ThalovantClient, build_client_context

context = build_client_context(
    user_id="user-42",
    user_name="Ada",
    auth_provider="oidc",
    roles=["member"],
    platform="kiosk",
    source="checkout-kiosk",
    channel="chat",
)

with ThalovantClient.from_identity_file("_identity.json") as client:
    reply = client.ask("Show the next instruction.", context=context)
    print(reply.text)

Actions And Exact Inputs

Use actions for button payloads and codes for exact typed or scanned values.

with client.conversation(session_id="work-session") as convo:
    convo.send_action('/choose{"id":"42"}', title="Choose item")
    convo.send_code("SN-001-XYZ", kind="qr", label="serial")

Rich Responses

Replies can include text, choices, tables, images, or attachments.

reply = client.ask("Show matching parts.")

for item in reply.display_items(max_text_chars=600):
    if item.kind == "text":
        print(item.text)
    elif item.kind == "choices":
        print([choice["title"] for choice in item.data])

Async Apps

import asyncio
from thalovant import AsyncThalovantClient


async def main():
    async with AsyncThalovantClient.from_identity_file("_identity.json") as client:
        reply = await client.ask("What time is it?")
        print(reply.text)


asyncio.run(main())

CLI Diagnostics

thalovant --identity _identity.json doctor

The doctor command checks identity shape, endpoint selection, authentication, handshake, and transport health.

Common Issues

• Missing Thalovant API access token: call api.login(...) before private control-plane actions, or pass access_token= to ThalovantControlPlane.
• API access requires a paid plan: upgrade the workspace before using the SDK control-plane API to provision private resources.
• Unsupported protocol: the hub does not expose that protocol, or the identity was created before that protocol was enabled.
• MQTT fails immediately: create or download a fresh client identity after MQTT is enabled. MQTT needs the per-client identity.mqtt credentials.
• A request times out: increase timeout on ask(...) or check doctor().

Development

pip install -e ".[dev]"
pytest