Table of contents
- Overview
- Installation
- Architecture concepts
- Module: Log
- Module: RemoteSession
- Data classes
- Class RemoteSession
- Module: RemoteSessionPool
- Module: PlanManager
- Class PlanFragment
- Class PlanManager
- End-to-end examples
- Single-session question and answer
- Batch processing with a session pool
- Managing custom plans
Overview
The thalabus Python SDK lets you write programs that talk to a running thalabus server as if it were a conversational AI service you control completely. Everything from starting a session, sending a question, waiting for the LLM to finish reasoning, reading the structured answer, and tearing the session down again is a straightforward await call.
What the SDK is for
- Batch automation — process hundreds of documents, one per session, in a managed pool of parallel connections.
- Programmatic prompting — send prompts, system messages, and file attachments to a thalabus session from Python code rather than a browser.
- Structured extraction — instruct the LLM to return a specific JSON shape; the SDK automatically deserialises the response for you.
- Plan management — create, read, update, and delete custom execution plans (PlanFragments) on the server via the agent API.
- Integration testing — spin up a session, assert over the response, and tear it down, all inside a test suite.
Package layout
| Module | Main class | Purpose |
|---|---|---|
thalabus.Log |
— | Structured logging with levels and file output |
thalabus.RemoteSession |
RemoteSession |
Single connection to one thalabus session |
thalabus.RemoteSessionPool |
RemoteSessionPool |
Parallel pool of sessions for batch work |
thalabus.PlanManager |
PlanManager |
CRUD over custom plan fragments on the server |
Installation
Install from PyPI using pip:
pip install thalabus
Or install from a local checkout:
cd ThalabusSDK
pip install -e .
Python requirement: 3.9 or later.
Runtime dependencies (installed automatically by pip):
| Package | Purpose |
|---|---|
aiohttp |
Async HTTP client used for all server communication |
Architecture concepts
Before reading the API reference it helps to know the four building blocks of the thalabus platform that the SDK exposes.
Sessions
A session is the unit of conversation on the thalabus server. One session equals one chatbot conversation. Each session has:
- a unique string id (e.g.
"sdk-1","myapp-batch-003") - an owner user id
- a plan — the goal-tree describing what the session is meant to accomplish
- an active LLM configuration — which models are used and at what settings
- a per-session API token that your code must pass for all subsequent calls
Sessions are created (or attached to if they already exist) by calling RemoteSession.create(), and removed by calling RemoteSession.destroy().
Plans and goals
A plan is a structured tree of goals that the AI agents in the session execute. The SDK exposes the current plan through RemoteSession.get_plan() and through the RemoteSession.plan property. You can also recommend a specific named plan at prompt-submission time via the recommended_plan parameter of submit_message().
LLM configurations
A named configuration binds each agent role inside a session to a particular language model and its sampling parameters. Multiple configurations can coexist on one server (e.g. "default", "fast", "gpt-4o"). You switch the active one per session with set_active_configuration().
Custom plans (PlanFragments)
A PlanFragment is a reusable, named template that encapsulates one or more agent prompts. Custom plan fragments are user-authored and managed through the PlanManager class.
Module: Log
Import path: thalabus.Log
The log module provides a thread-safe, colour-coded logging system that writes simultaneously to the console and to a file. Always call log_init() at the start of your program and log_exit() at the end (typically in a finally block).
Log levels
Log levels are integer constants. Each level also implies all higher levels when filtering.
| Constant | Value | Colour | Meaning |
|---|---|---|---|
DEBUG |
0 | Gray | Low-level tracing, verbose inner detail |
LLM_IN |
1 | Magenta | Prompt text sent to the LLM |
LLM_OUT |
2 | Blue | Raw response received from the LLM |
FUNCTION |
3 | Green | Python function call/return events |
INFO |
4 | White | Normal informational messages |
PLAN |
5 | Black | Plan and goal state changes |
WARNING |
6 | Yellow | Recoverable problems |
ERROR |
7 | Red | Non-fatal failures |
FATAL |
8 | Light red | Unrecoverable failures |
Functions
log_init(log_filename, disable_screen_lock, limit_console_log_level)
Initialises the logging system. Must be called once before any log() call. Calling it multiple times is safe — subsequent calls are no-ops.
| Parameter | Type | Default | Description |
|---|---|---|---|
log_filename |
str |
"thalabus.log" |
Path of the log file to create (relative to the working directory, or /app/ inside Docker) |
disable_screen_lock |
bool |
False |
Set True to skip the per-line mutex on console output. Useful in single-threaded scripts where the lock adds overhead. |
limit_console_log_level |
int or None |
None |
When set, messages below this level are not printed to the console (but they are still written to the file). For example, pass INFO to silence all DEBUG, LLM_IN, LLM_OUT, and FUNCTION noise on screen. |
Example:
from thalabus.Log import log_init, log_exit, log, INFO, ERROR
log_init() # default: writes to ./thalabus.log, all levels
# or:
log_init(limit_console_log_level=INFO) # only INFO and above on the console
log_exit()
Flushes and closes the log file, and waits for the background flush thread to terminate. Call this in a finally block so it always runs even if an exception propagates.
try:
# your code
pass
finally:
log_exit()
log(level, msg)
Emits a single log line. The line is written to both the console (with ANSI colour) and the log file (plain text). The line is automatically prefixed with the current timestamp, log level string, calling module name, calling function name, and (when available) the session id.
| Parameter | Type | Description |
|---|---|---|
level |
int |
One of the level constants (DEBUG, INFO, ERROR, etc.) |
msg |
str or dict or list |
The message. Dicts and lists are pretty-printed as formatted JSON. |
from thalabus.Log import log, INFO, ERROR, DEBUG
log(INFO, "Processing file: report.txt")
log(DEBUG, {"key": "value", "nested": [1, 2, 3]}) # pretty-printed
log(ERROR, "Connection refused")
Do not split a log message across multiple calls to achieve multi-line output. Pass the full string, including embedded newlines, in a single log() call.
log_plan(plan)
Writes the full contents of a RemotePlan object to the plan.log file. Useful for debugging the current goal-tree without cluttering the main log.
plan = await remote_session.get_plan()
log_plan(plan)
Module: RemoteSession
Import path: thalabus.RemoteSession
This module defines the two data-holder classes (ContainerMessage, Session) and the main RemoteSession class which wraps a single connection to a thalabus session over HTTP.
All methods that touch the network are async — call them with await inside an async function or via asyncio.run().
Data classes
ContainerMessage
A simple result envelope returned by most RemoteSession methods to communicate either success or failure.
| Attribute | Type | Description |
|---|---|---|
msg_message |
str or None |
The payload of the message. On success this is a human-readable string (or the session id). On failure this is the error description. |
msg_is_error |
bool |
True when the operation failed. Always check this before using msg_message. |
Example:
msg = await rs.create(endpoint, session_id, token)
if msg.msg_is_error:
log(ERROR, f"Could not create session: {msg.msg_message}")
else:
log(INFO, f"Session ready: {msg.msg_message}")
RemotePlan
A snapshot of the session's current execution plan as returned by the server.
| Attribute | Type | Description |
|---|---|---|
plan_name |
str |
Name of the active plan |
plan_description |
str |
Human-readable description |
plan_domain |
str |
Domain category of the plan |
plan_focus |
str |
Focus area |
plan_default_language |
str |
BCP-47 language code (e.g. "en", "fr") |
plan_language_in_use |
str |
Currently active language |
plan_document_ids |
list |
IDs of documents loaded into this session |
plan_loaded_documents |
list or None |
Full document objects if the server returned them |
plan_datasets |
list |
Dataset identifiers attached to this session |
plan_revision |
str |
Revision token used for conditional GET (returns 304 Not Modified when unchanged) |
You normally do not need to instantiate this class yourself — it is returned by get_plan() and exposed via the plan property.
Session
Internal object that holds the raw data for one connected session. You rarely interact with it directly; use the RemoteSession properties instead.
| Attribute | Type | Description |
|---|---|---|
s_id |
str |
Session id |
s_title |
str |
Human-readable title |
s_endpoint_base |
str |
The /v1 base URL in use |
s_userId |
str |
Owner user id |
s_plan |
RemotePlan |
Cached plan object |
s_activeLlmConfigurationName |
str |
Name of the active LLM configuration |
s_api_token |
str |
Per-session API token |
Class RemoteSession
RemoteSession manages one logical connection to one session on the thalabus server. It knows the server endpoint, holds the session credentials after connect/create, and provides methods for every operation you need.
Construction and configuration
RemoteSession(endpoint=None, user_id="test")
Creates a new RemoteSession object. At this point no network call is made; the object is just configured.
| Parameter | Type | Default | Description |
|---|---|---|---|
endpoint |
str or None |
None |
Base URL including the /v1 path suffix (e.g. "http://localhost:5210/v1"). When None, the class defaults to http://localhost:5210/v1. |
user_id |
str |
"test" |
The user id that will own newly created sessions. Not used when connecting to an existing session. |
Note: The string "localhost" in the endpoint is automatically replaced by the class-level BLAZOR_HOSTNAME constant, which defaults to "localhost". In containerised deployments override the constant before instantiating, or pass the endpoint explicitly.
from thalabus.RemoteSession import RemoteSession
rs = RemoteSession(endpoint="http://192.168.30.41:8080/v1", user_id="myapp")
Session lifecycle
await rs.create(endpoint_base, session_id, token) → ContainerMessage
Ensures that a session with the given id exists on the server, creating it if it does not, and attaches to it. This is the method to use in batch automation where you manage session lifecycle yourself.
Internally the method first pings the server to verify it is reachable, then attempts to GET the session. If the session does not exist (any non-200 response on the first attempt), it issues a POST to create the session, waits 2 seconds, and re-fetches.
| Parameter | Type | Description |
|---|---|---|
endpoint_base |
str |
Full base URL including /v1 (e.g. "http://localhost:5210/v1") |
session_id |
str |
Desired session id. May already exist; if so, the method attaches to it. Recommended convention: use a stable prefix plus a numeric suffix, e.g. "batch-001". |
token |
str |
Platform-level API token. Needed only for creation; after connecting the per-session token in rs.api_token is used for subsequent calls. |
Returns: ContainerMessage where msg_is_error is False on success and msg_message is the resolved session id.
msg = await rs.create(
endpoint_base="http://localhost:5210/v1",
session_id="my-session-1",
token="1234-XXXX-87654321"
)
if msg.msg_is_error:
raise RuntimeError(msg.msg_message)
await rs.connect(endpoint_base, session_id, token) → ContainerMessage
Attaches to a session that must already exist. Unlike create(), this method does not attempt to create the session if it is absent — the call fails immediately with an error.
Use connect() when you know the session was created externally (e.g. the user already has an open browser tab) and you only want to observe or interact with it.
| Parameter | Type | Description |
|---|---|---|
endpoint_base |
str |
Full base URL including /v1 |
session_id |
str |
Id of the existing session |
token |
str |
Platform-level API token |
Returns: ContainerMessage — same contract as create().
rs.disconnect() → None
Detaches the RemoteSession object from its session without deleting the session on the server. After calling disconnect() the session continues to exist and run on the server; you just drop your local reference to it.
This is useful when you hand off a session to another process or simply want to stop interacting with it while keeping it alive for the user.
rs.disconnect()
# rs.id is now None; rs.rs_session is None
await rs.destroy() → ContainerMessage
Sends a DELETE request to the server to permanently remove the session, then disconnects locally. Any conversation history, loaded documents, and plan state are lost.
Use destroy() at the end of a batch job when the sessions are no longer needed and you want to free server resources.
Returns: ContainerMessage — msg_is_error is False on success.
msg = await rs.destroy()
if msg.msg_is_error:
log(WARNING, f"Destroy returned an error: {msg.msg_message}")
await rs.clear(timeout=600) → None
Sends the special /clear command to the session, which resets the conversation history (chat lines) and the current plan, leaving the session alive and ready for a new task. Internally it waits for the session to be idle before sending the command, then waits again for the server to acknowledge.
| Parameter | Type | Default | Description |
|---|---|---|---|
timeout |
int |
600 |
Maximum seconds to wait for the session to become idle before and after the clear command |
The RemoteSessionPool calls clear() automatically before dispatching each task to a session, so you do not need to call it yourself in pool-based code.
Sending content
await rs.submit_prompt(prompt) → None
Sends a plain user-role text message to the session. Equivalent to the user typing prompt into the chat interface.
| Parameter | Type | Description |
|---|---|---|
prompt |
str |
The text to send |
await rs.submit_prompt("What are the main risks described in this document?")
await rs.submit_message(prompt, role, json_output, structured_output, timeout_sec, recommended_plan, recommended_tools) → None
The full-featured message submission method. submit_prompt() is a thin wrapper around this.
| Parameter | Type | Default | Description |
|---|---|---|---|
prompt |
str |
— | The message text |
role |
str |
"user" |
Message role: "user" or "system" |
json_output |
dict or None |
None |
When provided, instructs the LLM to return a JSON object whose keys match this dictionary. The keys serve as field names and the string values serve as field descriptions. See Structured extraction below. |
structured_output |
any |
None |
Alternative structured output hint, passed through to the server as-is |
timeout_sec |
int |
600 |
Seconds to wait for the session to become idle before submitting. The method calls wait_until_ready() first. |
recommended_plan |
str or None |
None |
Name of a plan fragment to recommend. When provided, the server prefers this plan over auto-selection. Use plan names like "plan_simple_answer" to skip reasoning steps you do not need. |
recommended_tools |
list[str] or None |
None |
Optional list of tool names to recommend the LLM use |
Structured extraction
Pass a dict to json_output where each key is a field you want in the response and the value is a description:
await rs.submit_message(
"Extract the key metadata from the attached article.",
json_output={
"title": "The full title of the article",
"authors": "A list of author names",
"publication_date": "Date in DD.MM.YYYY format",
"abstract": "A one-paragraph summary"
},
recommended_plan="plan_simple_answer"
)
After sending this, call get_response() — it will automatically JSON-decode the LLM's reply if the reply looks like a JSON object or array.
await rs.submit_attachment(attachment) → None
Sends the content of attachment as a system-role message, framed as "the following text is attached to the conversation". This is how you inject the text of a document or file into the session for the LLM to analyse.
| Parameter | Type | Description |
|---|---|---|
attachment |
str |
The full text content to attach |
with open("report.txt", "r", encoding="utf-8") as f:
text = f.read()
await rs.submit_attachment(text)
# Now send the actual question:
await rs.submit_prompt("Summarise the key findings from this report.")
Polling and reading responses
await rs.wait_until_ready(timeout_sec=600) → ContainerMessage
Blocks until the session transitions to "awaiting input" state (i.e. all agents have finished processing and the session is ready to receive the next message). This is a long-poll: the server holds the connection open until either the session becomes ready or timeout_sec elapses.
| Parameter | Type | Default | Description |
|---|---|---|---|
timeout_sec |
int |
600 |
How many seconds the server should wait before timing out. Pass 0 to do a non-blocking check (returns immediately with the current state). |
Returns: ContainerMessage — msg_is_error is False when the session is ready. The msg_message field is a human-readable status string such as "Awaiting input: True".
status = await rs.wait_until_ready(timeout_sec=120)
if status.msg_is_error:
raise RuntimeError("Session never became ready")
await rs.is_awaiting_input() → ContainerMessage
Non-blocking version of wait_until_ready(). Immediately returns the current readiness state without any server-side waiting. Equivalent to wait_until_ready(0).
status = await rs.is_awaiting_input()
if not status.msg_is_error:
log(INFO, "Session is idle and ready")
await rs.wait_for_response(timeout_sec=600, is_satisfied=None) → list or None
Waits until the session is ready, fetches the chat lines, filters out internally suppressed messages, and optionally checks whether the response meets a custom criterion. If the criterion fails (e.g. the last message is still an intermediate function-result line), the method loops: it calls wait_until_ready() again and retries.
| Parameter | Type | Default | Description |
|---|---|---|---|
timeout_sec |
int |
600 |
Passed to wait_until_ready() on each iteration |
is_satisfied |
callable or None |
None |
Optional predicate (visible_lines: list) → bool. When None, the first ready state is accepted. |
Returns: The list of visible (non-suppressed) chat line dicts, or None on error. Each dict has at least a "message" key with the text content and a "role" key.
def final_answer_ready(lines):
# only accept when the last visible message is from the assistant
return lines and lines[-1].get("role") == "assistant"
lines = await rs.wait_for_response(is_satisfied=final_answer_ready)
if lines:
print(lines[-1]["message"])
await rs.get_response(timeout_sec=600, retries=5, retry_delay_sec=30) → dict or str or None
High-level convenience method: waits for the session to be ready, fetches chat lines, finds the last visible (non-suppressed) message, and attempts to JSON-decode it. Returns a dict or list if the message is valid JSON, otherwise returns the raw string.
| Parameter | Type | Default | Description |
|---|---|---|---|
timeout_sec |
int |
600 |
Passed to wait_until_ready() |
retries |
int |
5 |
How many times to retry if no visible message is found yet |
retry_delay_sec |
int |
30 |
Seconds to sleep between retries |
Returns: dict (JSON object), list (JSON array), str (raw text), or None if nothing arrived within the retry budget.
response = await rs.get_response()
if isinstance(response, dict):
print(response["title"]) # structured JSON
elif isinstance(response, str):
print(response) # plain text answer
else:
log(ERROR, "No response received")
await rs.get_chat_lines(skillset_id=None) → list or None
Returns the full, unfiltered list of chat lines for this session. Each element is a dict with fields such as "message", "role", and "goalMessageSuppress". Unlike get_response(), this method does not wait for the session to become ready — call wait_until_ready() first if you need to ensure the LLM has finished.
await rs.wait_until_ready()
lines = await rs.get_chat_lines()
for line in lines:
print(f"[{line.get('role', '?')}] {line.get('message', '')[:120]}")
await rs.get_plan() → RemotePlan or None
Fetches the current plan from the server. Uses the plan revision stored in the session to issue a conditional request — if the plan has not changed since the last fetch, the server returns 304 Not Modified and the cached plan is returned without network overhead.
Returns: RemotePlan on success, None on error.
plan = await rs.get_plan()
if plan:
print(f"Active plan: {plan.plan_name}")
print(f"Language: {plan.plan_language_in_use}")
print(f"Documents: {plan.plan_document_ids}")
LLM configuration
await rs.set_active_configuration(configuration_name) → ContainerMessage
Switches the active LLM configuration for this session. The named configuration must already exist on the server (defined in the server's LLM settings file or via the admin interface).
| Parameter | Type | Description |
|---|---|---|
configuration_name |
str |
Name of the configuration to activate (e.g. "fast", "gpt-4o") |
msg = await rs.set_active_configuration("fast")
if msg.msg_is_error:
log(WARNING, f"Could not switch configuration: {msg.msg_message}")
await rs.get_active_configuration() → str
Returns the name of the currently active LLM configuration for this session, or an empty string if the request fails.
config_name = await rs.get_active_configuration()
print(f"Using: {config_name}")
Utility methods
await rs.ping() → None
Issues a lightweight call (get_active_configuration) to verify that the connection is still alive. Useful as a keep-alive probe in long-running programs.
rs.validate_token(token) → bool
Returns True if token equals the session's own api_token. Useful when your code needs to verify that a token it holds is still the correct one for this session.
Properties
All properties below raise no exception — they return None if the session is not yet connected.
| Property | Type | Description |
|---|---|---|
rs.id |
str or None |
Session id, e.g. "my-session-1" |
rs.title |
str or None |
Human-readable session title |
rs.user_id |
str or None |
Owner user id |
rs.plan |
RemotePlan or None |
Cached plan snapshot (updated by get_plan()) |
rs.active_llm_configuration_name |
str or None |
Name of the active LLM configuration as reported at connect/create time |
rs.api_token |
str or None |
Per-session API token for authenticating subsequent requests |
Module: RemoteSessionPool
Import path: thalabus.RemoteSessionPool
RemoteSessionPool manages a fixed-size pool of RemoteSession objects and dispatches a list of tasks to them in parallel. Sessions are created upfront, reused across tasks (with a clear() call between tasks), and optionally destroyed when all tasks complete.
This is the right tool whenever you have many independent items to process (files, records, queries) and want to run them concurrently against a thalabus server.
RemoteSessionPool(session_count, endpoint, token, session_prefix, user_id)
Creates and configures the pool. No sessions are created at this point; that happens lazily inside pool_execute().
| Parameter | Type | Description |
|---|---|---|
session_count |
int |
Number of parallel sessions to maintain. Tune this based on the server's capacity. Typical values: 2–10. |
endpoint |
str |
Full /v1 base URL of the thalabus server |
token |
str |
Platform-level API token used to create sessions |
session_prefix |
str |
String prepended to a 1-based index to form each session id. E.g. "batch-" yields "batch-1", "batch-2", … |
user_id |
str |
Owner user id assigned to created sessions |
from thalabus.RemoteSessionPool import RemoteSessionPool
pool = RemoteSessionPool(
session_count=5,
endpoint="http://localhost:5210/v1",
token="1234-XXXX-87654321",
session_prefix="sdk-",
user_id="myapp"
)
await pool.pool_execute(task, task_args, keep_alive=False) → list
The main entry point. Creates sessions in the background, dispatches each element of task_args to the first available session by calling your task coroutine, waits for all tasks to finish, and optionally destroys the sessions.
| Parameter | Type | Description |
|---|---|---|
task |
async callable |
Your coroutine with signature async def task(rs: RemoteSession, task_arg: any). Called once per element of task_args. |
task_args |
list |
One element per unit of work. Elements are passed unchanged as the second argument to task. |
keep_alive |
bool |
When False (default), all sessions are destroyed after the last task completes, freeing server resources. When True, sessions remain alive — useful for inspection or when you plan another pool_execute() call. |
Returns: A list of results (the return values of all task calls). In practice most tasks do their own I/O (write files, append to a list, etc.) and return None.
How the pool schedules work
- Session creation runs concurrently with task dispatch: as soon as the first session is ready, the first task is dispatched to it, even if other sessions are still being created.
- Each task calls
wait_until_ready()before sending anything, then callsclear()to reset the conversation. - When a task finishes (or raises an exception), the session is returned to the free pool immediately so the next waiting task can use it.
- If a task raises an exception, that exception is logged and the session is still returned to the pool — other tasks are not affected.
Your task function should be self-contained. Do not share mutable state between task calls unless protected by a lock, because tasks run concurrently.
Module: PlanManager
Import path: thalabus.PlanManager
PlanManager provides create, read, update, and delete operations for custom plan fragments — user-authored execution templates stored on the thalabus server under /v1/agent/customPlans.
Class PlanFragment
Represents one plan fragment as returned by the server.
| Attribute | Type | Description |
|---|---|---|
id |
str |
Server-assigned unique identifier |
name |
str |
Human-readable name shown in the UI |
description |
str |
Free-text description of what this plan does |
blacklisted |
bool |
When True, this plan is hidden from the user's plan picker |
is_custom_plan |
bool |
True for user-authored plans; False for built-in server plans |
template_agents |
list[dict] |
List of agent definitions. Each dict has at least a "prompt" key containing the system prompt for that agent. |
Property:
| Property | Type | Description |
|---|---|---|
first_agent_system_prompt |
str |
Returns the system prompt of the first agent in the plan, or "" if there are no agents. Convenient shortcut for inspecting single-agent plans. |
Method:
| Method | Returns | Description |
|---|---|---|
to_dict() |
dict |
Serialises the fragment back to a dict suitable for passing to create_custom_plan() or update_custom_plan(). |
Class PlanManager
PlanManager(endpoint_base, token, username)
| Parameter | Type | Description |
|---|---|---|
endpoint_base |
str |
Full /v1 base URL (e.g. "http://localhost:5210/v1") |
token |
str |
Platform-level API token |
username |
str |
Username scoping the plan operations (plans can be user-specific) |
from thalabus.PlanManager import PlanManager
pm = PlanManager(
endpoint_base="http://localhost:5210/v1",
token="1234-XXXX-87654321",
username="alice"
)
await pm.get_custom_plans() → list[PlanFragment]
Returns all custom plans visible to the configured username. Returns an empty list on error.
plans = await pm.get_custom_plans()
for p in plans:
print(f"{p.id:36} {p.name}")
await pm.get_custom_plan(plan_id) → PlanFragment or None
Returns a single plan fragment by its id, or None if not found or on error.
plan = await pm.get_custom_plan("abc-123")
if plan:
print(plan.first_agent_system_prompt)
await pm.create_custom_plan(name, description, template_agents) → PlanFragment or None
Creates a new custom plan on the server.
| Parameter | Type | Default | Description |
|---|---|---|---|
name |
str |
— | Display name of the new plan |
description |
str |
"" |
Optional description |
template_agents |
list[dict] |
[] |
List of agent definitions. Each dict should have a "prompt" key at minimum. |
Returns: The newly created PlanFragment with its server-assigned id, or None on failure.
new_plan = await pm.create_custom_plan(
name="Extract Legal Entities",
description="Extracts all named legal entities (persons, organisations) from the attached document.",
template_agents=[{
"prompt": (
"You are a legal analyst. Extract all named persons and organisations "
"from the user's document. Return a JSON object with keys 'persons' "
"and 'organisations', each a list of strings."
)
}]
)
if new_plan:
print(f"Created plan: {new_plan.id}")
await pm.update_custom_plan(plan_id, name, description, blacklisted, template_agents) → PlanFragment or None
Updates an existing plan. Any parameter left as None is left unchanged (the method fetches the current plan first to preserve existing values).
| Parameter | Type | Default | Description |
|---|---|---|---|
plan_id |
str |
— | Id of the plan to update |
name |
str or None |
None |
New name, or None to keep current |
description |
str or None |
None |
New description, or None to keep current |
blacklisted |
bool or None |
None |
Set True to hide the plan, False to show it, or None to keep current |
template_agents |
list[dict] or None |
None |
Replacement agent list, or None to keep current |
Returns: The updated PlanFragment, or None on failure.
updated = await pm.update_custom_plan(
plan_id="abc-123",
description="Updated: also extracts dates from documents."
)
await pm.delete_custom_plan(plan_id) → bool
Permanently deletes the plan with the given id.
Returns: True on success, False on failure.
ok = await pm.delete_custom_plan("abc-123")
if not ok:
log(ERROR, "Failed to delete plan")
End-to-end examples
Example 1: Single-session question and answer
This is the simplest possible usage. The program connects to an existing session, sends one question, waits for the response, prints it, and disconnects.
import asyncio
from thalabus.RemoteSession import RemoteSession
from thalabus.Log import log_init, log_exit, log, INFO, ERROR
ENDPOINT = "http://localhost:5210/v1"
SESSION_ID = "my-session-1"
TOKEN = "1234-XXXX-87654321"
async def main():
rs = RemoteSession(endpoint=ENDPOINT, user_id="demo")
# Connect to the session (it must already exist on the server)
msg = await rs.connect(ENDPOINT, SESSION_ID, TOKEN)
if msg.msg_is_error:
log(ERROR, f"Could not connect: {msg.msg_message}")
return
log(INFO, f"Connected to session: {rs.id}")
# Send a question
await rs.submit_prompt("What is the capital of France?")
# Wait for the answer
response = await rs.get_response()
log(INFO, f"Answer: {response}")
# Disconnect without deleting the session on the server
rs.disconnect()
if __name__ == "__main__":
log_init()
try:
asyncio.run(main())
finally:
log_exit()
Example 2: Batch processing with a session pool
This example processes every .txt file in a folder in parallel, using up to 5 concurrent sessions. For each file it attaches the file content and asks the LLM to extract structured metadata as JSON.
import asyncio
import fnmatch
import json
import os
from typing import List
from thalabus.RemoteSession import RemoteSession, ContainerMessage
from thalabus.RemoteSessionPool import RemoteSessionPool
from thalabus.Log import log_init, log_exit, log, INFO, DEBUG, ERROR, FATAL
FOLDER = "./documents"
PATTERN = "*.txt"
ENDPOINT = "http://localhost:5210/v1"
TOKEN = "1234-XXXX-87654321"
USER_ID = "batch-runner"
# Describe the JSON shape you want back from the LLM
JSON_OUTPUT = {
"title": "The full title of the document",
"summary": "A two-sentence summary of the document",
"key_entities": "A list of named persons or organisations mentioned",
"document_date": "The document date in DD.MM.YYYY format, or null if not found"
}
async def task(rs: RemoteSession, task_arg: dict):
"""Process one file. Called once per file by the pool."""
filename = task_arg["file"]
log(INFO, f"Processing: {filename}")
# Read and attach the file
with open(filename, "r", encoding="utf-8") as f:
text = f.read()
await rs.submit_attachment(text)
# Ask for structured extraction
await rs.submit_message(
"Extract the required metadata from the attached document. "
"Reply only with the JSON structure.",
json_output=JSON_OUTPUT,
recommended_plan="plan_simple_answer" # skip unnecessary reasoning steps
)
# Retrieve and save the response
response = await rs.get_response()
if response is None:
log(ERROR, f"No response for: {filename}")
return
out_path = os.path.splitext(filename)[0] + ".json"
with open(out_path, "w", encoding="utf-8") as f:
if isinstance(response, (dict, list)):
json.dump(response, f, indent=2, ensure_ascii=False)
else:
f.write(str(response))
log(INFO, f"Saved: {out_path}")
def find_files(folder: str, pattern: str) -> List[str]:
results = []
for root, _, filenames in os.walk(folder):
for name in fnmatch.filter(filenames, pattern):
results.append(os.path.join(root, name))
return results
async def main():
files = find_files(FOLDER, PATTERN)
if not files:
log(INFO, "No files found.")
return
log(INFO, f"Found {len(files)} files — dispatching to pool of 5 sessions")
pool = RemoteSessionPool(
session_count=5,
endpoint=ENDPOINT,
token=TOKEN,
session_prefix="batch-",
user_id=USER_ID
)
task_args = [{"file": f} for f in files]
await pool.pool_execute(task, task_args, keep_alive=False)
log(INFO, "All tasks complete.")
if __name__ == "__main__":
log_init()
try:
asyncio.run(main())
except Exception as e:
import traceback
log(FATAL, f"Unhandled exception: {e}\n{traceback.format_exc()}")
finally:
log_exit()
Example 3: Managing custom plans
This example shows how to list all custom plans, create a new one, update its description, and then delete it.
import asyncio
from thalabus.PlanManager import PlanManager
from thalabus.Log import log_init, log_exit, log, INFO, ERROR
ENDPOINT = "http://localhost:5210/v1"
TOKEN = "1234-XXXX-87654321"
USERNAME = "alice"
async def main():
pm = PlanManager(endpoint_base=ENDPOINT, token=TOKEN, username=USERNAME)
# List all custom plans
plans = await pm.get_custom_plans()
log(INFO, f"Found {len(plans)} custom plans:")
for p in plans:
log(INFO, f" {p.id} {p.name!r}")
# Create a new plan
new_plan = await pm.create_custom_plan(
name="Risk Extractor",
description="Identifies and categorises risks from legal or compliance documents.",
template_agents=[{
"prompt": (
"You are a compliance analyst. Read the attached document carefully. "
"Identify every risk mentioned, classify it as Legal, Financial, or Operational, "
"and estimate its severity as High, Medium, or Low. "
"Return a JSON array where each element has keys: "
"risk_description, category, severity."
)
}]
)
if not new_plan:
log(ERROR, "Plan creation failed.")
return
log(INFO, f"Created plan: {new_plan.id!r} — {new_plan.name!r}")
# Update the description
updated = await pm.update_custom_plan(
plan_id=new_plan.id,
description="Identifies and categorises risks. Updated to also capture likelihood."
)
if updated:
log(INFO, f"Updated description: {updated.description!r}")
# Delete the plan
ok = await pm.delete_custom_plan(new_plan.id)
log(INFO, f"Deleted: {ok}")
if __name__ == "__main__":
log_init()
try:
asyncio.run(main())
finally:
log_exit()
Quick reference
RemoteSession method summary
| Method | Async | Description |
|---|---|---|
__init__(endpoint, user_id) |
No | Create object, no network call |
create(endpoint_base, session_id, token) |
Yes | Create or attach to a session |
connect(endpoint_base, session_id, token) |
Yes | Attach to an existing session |
disconnect() |
No | Drop local reference, keep server session |
destroy() |
Yes | Delete the session on the server |
clear(timeout) |
Yes | Reset chat history, keep session alive |
submit_prompt(prompt) |
Yes | Send a user-role message |
submit_message(...) |
Yes | Send message with full options |
submit_attachment(attachment) |
Yes | Attach text as system context |
wait_until_ready(timeout_sec) |
Yes | Block until session is idle |
is_awaiting_input() |
Yes | Non-blocking readiness check |
wait_for_response(timeout_sec, is_satisfied) |
Yes | Wait and return visible chat lines |
get_response(timeout_sec, retries, ...) |
Yes | Wait and return last LLM message |
get_chat_lines(skillset_id) |
Yes | Return all chat lines unfiltered |
get_plan() |
Yes | Fetch current plan from server |
set_active_configuration(name) |
Yes | Switch LLM configuration |
get_active_configuration() |
Yes | Get current LLM configuration name |
ping() |
Yes | Keep-alive probe |
validate_token(token) |
No | Check if token matches session token |
PlanManager method summary
| Method | Async | Description |
|---|---|---|
get_custom_plans() |
Yes | List all custom plans |
get_custom_plan(plan_id) |
Yes | Fetch one plan by id |
create_custom_plan(name, ...) |
Yes | Create a new plan |
update_custom_plan(plan_id, ...) |
Yes | Update fields of an existing plan |
delete_custom_plan(plan_id) |
Yes | Permanently delete a plan |
thalabus Python SDK — v1 — © 2026 t2b AG