SDK Python

Host packages and execution options

Both SDKs accept a module path or a pkg:organization.package/identifier selector. Declare the module in the installed package's hosts. Installation verifies content; activation is an explicit application choice.

Setting	JavaScript	Python
Attestation	humanApprovalHost	human_approval_host
Agent decisions	agentDecisionProvider	agent_decision_provider
Attestation timeout, ms	humanApprovalAuthorityTimeoutMs	human_approval_authority_timeout_ms
Pure-function detail	tracePureFunctions	trace_pure_functions

The timeout defaults to 5,000 ms, with a range of 1 to 60,000 ms. Detail defaults to full; calls keeps calls and governance with fewer internal events. See packages and traces.

The Python SDK calls the same governed CLI as the JavaScript SDK. Both guides follow the same path, with the same concepts and introductory examples.

Handler callback arguments

JavaScript calls run({ request, context }); Python calls run(request, context). The declared input signature determines request.

Declared inputs	JavaScript request	Python request
None	{}	{}
One: value as Text	"Invoice 42"	"Invoice 42"
Two: left as Int, right as Int	{ left: 2, right: 3 }	{"left": 2, "right": 3}

With one input, read the value directly from request. Context contains execution metadata and helpers; business inputs are never siblings of it. Without an explicit input signature, the handler receives the arguments map.

Installation

From your application root, choose the Python SDK:

touba sdk install python

Touba verifies its bundled wheel, copies it into vendor/touba-sdk/<hash>/, creates or reuses .venv, and installs touba-sdk offline. It checks both ToubaEngine and AsyncToubaEngine imports. Nothing is installed globally. Python 3.11+ is required, with uv or venv and pip.

Choose the package manager

touba sdk install python --manager uv
touba sdk install python --manager pip

The default auto selects uv when the project has a uv.lock, otherwise pip. Install the selected manager first. With uv and a pyproject.toml, Touba records the local wheel as a dependency and updates uv.lock offline. A later uv sync therefore retains the SDK. Without a declared Python project, Touba prepares only the virtual environment.

Offline resolution of an existing uv project requires its other dependencies to be available in the uv cache. Touba installs only its wheel; you choose when to sync the rest of the project. For Poetry, PDM or another manager, use the verified archive returned by touba sdk path python with your manager.

To select the base interpreter for a new venv:

touba sdk install python --python python3.12

An existing .venv keeps its interpreter. Activate it before running your app:

source .venv/bin/activate
python -c "from touba_sdk import ToubaEngine, AsyncToubaEngine; print('SDK ready')"

On Windows use .venv\Scripts\Activate.ps1, or invoke .venv\Scripts\python.exe directly without activation. Touba must remain on PATH, or pass its absolute path as ToubaEngine(command=...).

Commit vendor/touba-sdk with your dependency declarations, and exclude .venv. In uv projects, commit uv.lock as well.

On Windows, the SDK recognizes the installed touba.cmd launcher and calls the bundled Node executable with the CLI entry point, without a command shell. Paths with spaces and arguments remain literal values. Custom .cmd or .bat scripts are refused: use an executable with separate arguments. This discovery does not verify publisher signatures.

Updates

After updating Touba, explicitly update the project's SDK:

touba sdk install python

Touba retains the existing venv and verifies the new archive. Other projects' SDKs remain unchanged.

Imports

from touba_sdk import ToubaEngine, ToubaEngineError, ToubaHandlerError
from touba_sdk.testing import assert_run_succeeded

The installation package is touba-sdk; its import module is touba_sdk. Test helpers are in its testing submodule.

Run a program

program.tb

module sdk.engine
strategy Review
    takes value as Text
    gives Text
    do
        give value

run.py

from touba_sdk import ToubaEngine

engine = ToubaEngine(project_root=".")
run = engine.run(source="program.tb", inputs={"value": "Invoice 42"})
print(run.status, run.reason, run.result)

In an asynchronous application, use AsyncToubaEngine:

import asyncio
from touba_sdk import AsyncToubaEngine

async def main():
    engine = AsyncToubaEngine(project_root=".")
    run = await engine.run(source="program.tb", inputs={"value": "Invoice 42"})
    print(run.status, run.reason, run.result)

asyncio.run(main())

RunOptions provides the full typed option list. Python uses snake_case: entry_ref, manifest, secrets, inputs, resource_bindings, supply_chain_lock, require_supply_chain_lock, effect_journal, agent_decision_provider, run_limits, max_provider_calls, human_approval_host, human_decisions, user_inputs, human_checkpoint_out, human_resume_checkpoint, parallel_checkpoint_out, parallel_resume_checkpoint, trigger_event, trigger_state, skill_packs, active_skill_pack_refs. Relative paths use project_root.

Compile a program

compiled = engine.compile("program.tb")
run = engine.run(air_json=compiled.air_json, inputs={"value": "Invoice 42"})
# compiled.air_document is the parsed object; air_json also accepts this object.

This extends run.py. compile returns CompileResult with air_json (text) and air_document (object). Supply exactly one input to run: source, air (file), or air_json (text or dictionary). With AsyncToubaEngine, also await the call to compile.

Results and errors

pendingInteractions (JavaScript) and pending_interactions (Python) expose all currently pending questions as {interaction_ref, prompt} objects, or []. The singular field identifies the first question for user_input_pending. Answer using userInputs / user_inputs keyed by occurrence, then resume with parallelResumeCheckpoint / parallel_resume_checkpoint. Private continuation state is not part of this public list.

RunResult exposes status, reason, result, messages, pending_interaction and trace. Pauses, human refusals and explicit stops remain controlled outcomes. Technical failures and runtime refusals raise ToubaEngineError with phase, code and safe details, without copying raw stderr.

engine.version() identifies the CLI; touba_sdk_metadata() identifies SDK builds by their content hash.

Store execution traces

run = engine.run(source="program.tb", trace_store=".touba/traces.sqlite")

SQLite stores traces in .touba/traces.sqlite at the engine’s project root by default. The path above can be customized. Each segment remains available after the process exits; traceOut / trace_out also produces a JSON export. See trace retention for archival.

For a long trace, choose trace_mode="reference". Once the journal is finalized, the response contains the business result, trace_reference, and trace set to None. Read events in pages:

trace_store = ".touba/traces.sqlite"
run = engine.run(
    source="program.tb", trace_store=trace_store, trace_mode="reference",
)
after = None
while True:
    page = engine.read_trace(
        trace_store=trace_store, reference=run.trace_reference,
        after=after, limit=100,
    )
    for event in page["events"]:
        print(event["kind"])
    if page["done"]:
        break
    after = page["next"]

read_trace uses the read-only Core reader. limit accepts 1 to 1,000 events; event content in each page is capped at 1 MiB. Runtime errors carry their reference in error.details["trace_reference"]. References support reading; execution resumes from its checkpoint. Keep the reference and journal to resume reading after a restart. Start without a cursor to verify the full segment digest; subsequent pages check their events and order.

With AsyncToubaEngine, use await engine.run(...) and await engine.read_trace(...) with the same parameters.

In inline mode, trace contains the complete document. The 32 MiB transport limit applies to SDK responses, including business results and messages. A long trace can be stored and read in pages without increasing that limit. See the trace reference.

Write a handler

After declaring a tool, select Python and generate its handler:

touba init --language python
touba sdk install python
touba prepare --mode write

prepare generates the .handler.py file, its contract tests and a binding to .venv. Preserve generated constants and the define_tool call; implement run(request, context) with def or async def. One input becomes request directly; zero or multiple inputs form a named dictionary.

define_tool handles JSON process input, create_tool_handler adapts business code and handle_tool_request validates the envelope. Unexpected exceptions are redacted; use ToubaHandlerError for a secret-free public diagnostic. Reserve stdout for the protocol, not logs.

Follow the Python handlers tutorial.

Effects and provenance

Memory provenance uses an explicit envelope, separate from business data:

from touba_sdk import create_tool_result

return create_tool_result(record, memory_metadata=[
    {"memory_ref": "memory:app/Records", "operation": "read"},
])

This excerpt belongs inside a handler; record is the business value. A memory_metadata key in an ordinary dictionary does not create an envelope. The Core checks tool, scope and permission correspondence.

create_effect_receipt and create_compensation_receipt produce correlated receipts. Use context helpers create_effect_receipt and create_compensation_receipt after the actual effect. A receipt grants no permissions and does not replace external service idempotency.

Testing

from touba_sdk.testing import assert_run_succeeded, assert_trace_events

run = engine.run(source="program.tb", inputs={"value": "Invoice 42"})
assert_run_succeeded(run, "Invoice 42")

This extends run.py. touba_sdk.testing also provides create_scripted_decision_provider, assert_run_refused and assert_trace_events. A scripted provider proposes test decisions, never permissions. Generated contract tests check admission; add positive tests for business results and effects.

Resumes and parallelism

Resume with the operation's checkpoint, the same effect journal and a decision for the pending occurrence. human_checkpoint_out and human_resume_checkpoint track approvals; parallel_checkpoint_out and parallel_resume_checkpoint track parallel groups. The application attests identity through human_approval_host, never a model-provided dictionary.

AsyncToubaEngine supports concurrent calls through asyncio.gather. This does not replace a parallel group declared in Touba, which retains its own limits and resume rules. See parallel processing and approvals.

timeout=30 bounds transport; cancel_event=threading.Event() cancels a synchronous call. With asyncio, cancel the task. No automatic retry occurs. POSIX terminates the process group; Windows terminates the CLI without an equivalent descendant guarantee. An effect may already have happened: inspect the journal before resuming.

Monetary budgets

Import create_sqlite_provider_budget from touba_sdk.budget.sqlite. It shares the JavaScript SDK's SQLite format and refusal codes, with snake_case options and integer micro-units. Connecting model transports is a separate concern. See the provider cost guide and no-paid-call tutorial.

Additional capabilities

The Python Engine uses a fresh process per run. Model adapters, conversation history and MCP have Python APIs. Persistent Engine transport across multiple runs remains specific to the JavaScript SDK.

Providers and history

create_ai_sdk_decision_provider, create_open_ai_decision_provider and create_ollama_decision_provider follow the same decision contracts as their JavaScript counterparts. Call them with await provider(request). decision_interface="structured" is the default; "native" uses native function calls, still admitted by the Core before any effect.

from touba_sdk import create_open_ai_decision_provider
from touba_sdk.history.sqlite import create_sqlite_decision_history_store

store = create_sqlite_decision_history_store(filename="history.sqlite")
decide_agent_capability = create_open_ai_decision_provider(
    model="YOUR_MODEL_ID",
    decision_interface="native",
    history={"store": store, "namespace": "support"},
    max_output_tokens=512,
    max_retries=0,
)

This configures a provider without making a model call. Select an available model and a host-managed key. Pass the .py module through agent_decision_provider or --agent-decision-provider. The CLI selects the project's .venv; TOUBA_HOST_PYTHON selects an explicit interpreter path. Export decide_agent_capability or decide from the module. An approval module exports attest_human_approval and initiator_ref. These modules are trusted: the model never supplies their paths.

Adapters reuse the Node backend already included with Touba over private stdio. The wheel installs no additional provider SDK. The generic provider accepts Python generate_text, generate_object and json_schema callbacks following the AI SDK contract; raw Python vendor SDK messages/results need explicit adaptation.

Without a json_schema callback, the JSON schema is passed directly to the Python generation function. HTTP adapters also accept fetch and count_fetch: each callback receives {url, method, headers, body} and returns {status, headers, body}, with a text body and string-to-string headers. These exchanges are bounded to 8 MiB and remain private to the host transport. Do not log their headers, which may contain provider credentials.

RAM is the default store. Import create_memory_decision_history_store from touba_sdk for an explicit store. History options include max_bytes, max_messages, max_retained_bytes, max_summary_bytes, summarize and on_event. Structural compaction is local; only the optional host summarize callback produces a summary. Invocation IDs and native messages follow the same rules as JS. SQLite persists records; RAM does not survive module shutdown. Close directly created providers and stores with close().

OpenAI context options include context_window_tokens, safety_margin_tokens and timeout_ms. Context size and monetary spending are separate controls. When both are enabled, token-count requests need their own explicit tariff. A broken channel is never replayed automatically.

See the Python provider tutorial and shared history rules.

MCP

create_mcp_stdio_tool_bridge(mcp_tool_name=..., config=...) returns an async bridge(request, context) handler. Its touba-mcp-bridge-config-v1 format is identical to JavaScript. Python uses a bounded stdio subprocess directly, without another Node server inside the handler. Secret handles, time/size limits and refusals are preserved. Core admission still precedes tool execution.

Supply-chain locking follows static Python imports and package-manager lock files without executing handlers or auditing the installed venv. See supply chain. For a framework-based application, see Python orchestration.

Authority boundary

Both SDKs call the Touba CLI. The Core checks types, permissions, secrets, scopes, budgets, agent selection, approvals and effects before dispatch, then validates the returned result. The SDK grants no additional authority and is not a second language implementation.

Pass manifests, secrets and resource bindings explicitly. The Engine does not automatically load a .env. Protocol data and references are identical regardless of the SDK.

Switch between SDKs

Class names and concepts are shared. Functions and options use each language's usual casing. AIR, manifest and trace protocol keys remain unchanged.

JavaScript	Python
ToubaEngine, ToubaEngineError, ToubaHandlerError	ToubaEngine, ToubaEngineError, ToubaHandlerError
projectRoot, entryRef, pendingInteraction	project_root, entry_ref, pending_interaction
pendingInteractions	pending_interactions
compile({ source }) → airJson, airDocument	compile(source) → air_json, air_document
run({ airJson })	run(air_json=...)
defineTool, createToolHandler, handleToolRequest	define_tool, create_tool_handler, handle_tool_request
createToolResult, createEffectReceipt, createCompensationReceipt	create_tool_result, create_effect_receipt, create_compensation_receipt
toubaSdkMetadata	touba_sdk_metadata
@touba/sdk/testing	touba_sdk.testing
createScriptedDecisionProvider, assertRunSucceeded, assertRunRefused, assertTraceEvents	create_scripted_decision_provider, assert_run_succeeded, assert_run_refused, assert_trace_events

JavaScript uses await; Python provides synchronous ToubaEngine and AsyncToubaEngine with await. Business functions receive run({ request, context }) in JavaScript and run(request, context) in Python. These are calling conventions, not different responsibilities.

Follow the same steps in the JavaScript SDK, Python SDK, JavaScript Engine / Python Engine and JavaScript handlers / Python handlers tutorials.

Typed outputs and provider diagnostics

Date, format, length, min and max follow the same path in both SDKs: provider schema, response validation, then runtime validation. Invalid data is refused before use.

Setting	JavaScript	Python
Schema encoding	schemaEncoding	schema_encoding
Automatic selection (default)	"auto"	"auto"
Touba representation	"canonical"	"canonical"
Flattened optionals	"flat_optionals"	"flat_optionals"

Automatic mode adapts Optional fields for Anthropic models: it reduces unions in the transmitted schema, validates the response, then restores Touba's {present, value} encoding. Both validations stay active in structured and native interfaces. Malformed values are never converted to absence. Schema limits depend on the chosen model; qualify business types with that model.

In a host module, export initialModes (JS) or initial_modes (Python) to select summary or capabilities per agent reference:

export const initialModes = {
  "agent:app.invoices/Reader": "capabilities",
};

initial_modes = {"agent:app.invoices/Reader": "capabilities"}

Without an explicit setting, an agent with no capabilities starts directly with its result contract. When routing by agent, the SDK reports the resolved model; the agent_decision_provider_model_resolved event identifies the provider and model selected for the call.

Token-count failures retain the HTTP status and a bounded diagnostic. Interrupted generation exposes finish_reason and, when supplied, incomplete_reason. Partial results are never executed.

A write handler's public error code is preserved; missing or invalid receipts are reported separately. See External effects.

Route several providers

from models import providers

async def decide_agent_capability(request):
    provider = providers.get(request["agent_ref"])
    if provider is None:
        raise ValueError("No provider configured for this agent")
    return await provider(request)

providers contains adapters initialized once by the host module. The Python bridge propagates invocation context and costs through its private channel, including through this async wrapper. Keep the call in this context; the prompt contains the business request. The JavaScript router explicitly forwards its second execution argument.