Python SDK

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 calls clear() 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

Table of contents

Overview

What the SDK is for

Package layout

Installation

Architecture concepts

Sessions

Plans and goals

LLM configurations

Custom plans (PlanFragments)

Module: Log

Log levels

Functions

log_init(log_filename, disable_screen_lock, limit_console_log_level)

log_exit()

log(level, msg)

log_plan(plan)

Module: RemoteSession

Data classes

ContainerMessage

RemotePlan

Session

Class RemoteSession

Construction and configuration

RemoteSession(endpoint=None, user_id="test")

Session lifecycle

await rs.create(endpoint_base, session_id, token) → ContainerMessage

await rs.connect(endpoint_base, session_id, token) → ContainerMessage

rs.disconnect() → None

await rs.destroy() → ContainerMessage

await rs.clear(timeout=600) → None