titon-network-kronos-sdk 0.8.3

Python SDK for Kronos — decentralized automation for TON. Register recurring on-chain jobs, run automatons, decode events. TSA-audited zero findings.

titon-network-kronos-sdk

Python SDK for Kronos — decentralized automation for TON. Register recurring on-chain jobs, run automatons, decode events, react to slashes.

🛡️ TSA-audited — zero findings. Built on the same audited contracts as @titon-network/kronos-sdk on npm. See tsa-analysis/AUDIT-REPORT.md.

The Python SDK ships the same surface as the TypeScript SDK so docs and examples translate 1:1 — method names are snake_case mirrors of the TS camelCase (send_register_job ↔ sendRegisterJob, get_job ↔ getJob).

Install

pip install titon-network-kronos-sdk

Requires Python ≥ 3.11. The package depends on pytoniq-core (cells, addresses, BoC) and pytoniq (LiteClient, LiteBalancer, wallet abstractions).

Connect — testnet vs mainnet (the only difference)

The SDK is network-agnostic. You pick the network in two lines: the LiteServer config and which deployment constant you import.

import asyncio
from pytoniq import LiteBalancer
from kronos_sdk import KRONOS_TESTNET, KronosClient, KronosRegistry

async def main():
    # ── testnet ──────────────────────────────────────────────────────
    client = LiteBalancer.from_testnet_config(trust_level=2)
    await client.start_up()
    registry = KronosRegistry.create_from_address(KRONOS_TESTNET.registry, client=client)
    kronos = KronosClient(registry=registry)

    cfg = await kronos.jobs.config()
    print(f"connected to testnet — protocol fee {cfg.protocol_fee_bps} bps")
    await client.close_all()

asyncio.run(main())

For mainnet, change exactly two lines (once Kronos mainnet is deployed — see Live deployments below):

# ── mainnet ──────────────────────────────────────────────────────────
client = LiteBalancer.from_mainnet_config(trust_level=2)              # ← was from_testnet_config
await client.start_up()
registry = KronosRegistry.create_from_address(KRONOS_MAINNET.registry, # ← was KRONOS_TESTNET
                                              client=client)
kronos = KronosClient(registry=registry)

That's it. The same wheel, same bytecode, same code paths work on either network — only the LiteServer config and the deployment constant change. Pointing at a custom deployment (your own fork, a private chain) is just create_from_address("0Q...", client=client) — no constant required.

Three layers, pick what you need

1. Low-level wrapper — KronosRegistry

Matches the on-chain ABI 1:1; use directly when you need full control.

from pytoniq import LiteBalancer, WalletV4R2
from pytoniq_core import begin_cell

from kronos_sdk import KRONOS_TESTNET, KronosRegistry

client = LiteBalancer.from_testnet_config(trust_level=2)
await client.start_up()

registry = KronosRegistry.create_from_address(KRONOS_TESTNET.registry, client=client)
cfg = await registry.get_config()
print(f"protocol fee: {cfg.protocol_fee_bps} bps")

# Sending a job uses any pytoniq wallet
wallet = WalletV4R2.from_mnemonic(client, MNEMONIC.split())
await registry.send_register_job(
    wallet,
    value=int(1.05 * 10**9),  # 1.05 TON
    target=target_addr,
    message=begin_cell().store_uint(0x01, 32).end_cell(),
    interval=3600,
    reward=int(0.05 * 10**9),
    gas_limit=int(0.02 * 10**9),
    max_executions=100,
    window_before=30,
    window_after=600,
    expire_after=0,
)

2. Helpers — pure on-chain math

Mirror the contract's calculations off-chain; no network calls.

from kronos_sdk import (
    AssignmentInput, ExecutionEconomicsInput, JobWindowInput,
    assigned_automaton_index, execution_economics, job_window_state,
)

# Who runs the next execution?
idx = assigned_automaton_index(AssignmentInput(job_id=7, execution_count=3, active_automaton_count=4))
# → 2

# What does each execution cost?
e = execution_economics(ExecutionEconomicsInput(
    reward=int(0.05 * 10**9),
    gas_limit=int(0.02 * 10**9),
    protocol_fee_bps=500,
))
# e.total_cost = 0.0725 TON, e.protocol_fee = 0.0025 TON

# Is the job in its execution window?
state = job_window_state(JobWindowInput(
    last_executed_at=1_700_000_000,
    interval=3600,
    window_before=30,
    window_after=600,
    primary_window_seconds=30,
))
# state.status = 'primary' | 'fallback' | 'too-early' | 'too-late' | 'expired' | 'never-executed'

3. High-level — KronosClient, JobBuilder, decode_event

Path-of-least-resistance for most integrations.

from kronos_sdk import (
    KRONOS_TESTNET, KronosClient, KronosRegistry,
    JobBuilder, JOB_PRESETS, decode_event,
)
from pytoniq_core import begin_cell

registry = KronosRegistry.create_from_address(KRONOS_TESTNET.registry, client=client)
kronos = KronosClient(registry=registry)

# Namespaced reads
cfg = await kronos.jobs.config()
job = await kronos.jobs.get(job_id)
health = await kronos.jobs.balance_health(job_id)
if health.status == "low":
    print(f"job {job_id} only has {health.runs_remaining} runs left")

# Cross-contract: who's assigned to run this next?
assigned = await kronos.assigned_automaton_for(job_id)

# Fluent JobBuilder with safe defaults + auto-funding
opts = (JobBuilder()
    .target(target)
    .message(begin_cell().store_uint(my_op, 32).end_cell())
    .interval(3600)
    .reward(int(0.05 * 10**9))
    .gas_limit(int(0.02 * 10**9))
    .max_executions(1000)
    .with_preset(JOB_PRESETS["default"])
    .build(cfg))
await kronos.jobs.register(wallet, **opts.as_kwargs())

# Decode an external-out event body
ev = decode_event(external_out_body)
if ev and ev.kind == "JobExecuted":
    print(f"run {ev.execution_count} by {ev.automaton}, paid {ev.reward}")

API quick reference

Task	Call
Open the registry	KronosRegistry.create_from_address(addr, client=client)
High-level client	KronosClient(registry=registry) — exposes .jobs, .mirror, .events
Read config	await kronos.jobs.config() → RegistryConfigReply
Read job state	await kronos.jobs.get(job_id) → JobData | None
Health check	await kronos.jobs.balance_health(job_id) → JobBalanceHealth
Window state	await kronos.window_for(job_id) → JobWindowState
Who runs this next?	await kronos.assigned_automaton_for(job_id)
Mirror snapshot	await kronos.mirror.snapshot() → list[Address]
Send Execute	await kronos.jobs.execute(wallet, value=..., job_id=...)
Top up if low	await kronos.jobs.ensure_funded(wallet, job_id, EnsureFundedOptions(...))
Pure cost math	execution_economics(ExecutionEconomicsInput(...))
Pre-flight registration	register_job_opts(JobOptsInput(...), cfg) or JobBuilder().…build(cfg)
Decode an event body	decode_event(body) → typed KronosEvent | None
Subscribe per-job	JobWatcher(client, job_id, JobWatcherOptions(source=...))
Explain an exit code	explain_error(code) → ErrorExplanation
Live testnet addresses	KRONOS_TESTNET.registry / .forgeton
Bundled compiled BoC	load_registry_code() → Cell

For the AI-driven path (skills + system prompt), see AGENTS.md, AGENT-PROMPT.md, and skills/.

Live deployments

from kronos_sdk import KRONOS_TESTNET
KRONOS_TESTNET.registry      # → Address(...)
KRONOS_TESTNET.forgeton      # → Address(...) (the staking pool)
KRONOS_TESTNET.housekeeping_job_id

KRONOS_MAINNET lands once mainnet is live.

Error introspection

from kronos_sdk import explain_error, format_error_explanation, KronosError

e = explain_error(132)
# ErrorExplanation(code=132, origin='kronos', name='NotJobOwner',
#                  message='Operation requires the job owner.',
#                  related_skill='/kronos-job-lifecycle')

print(format_error_explanation(e))
# [NotJobOwner] (132) Operation requires the job owner.  See /kronos-job-lifecycle

# Wrap into an exception
raise KronosError(e)

Pool-side codes (160-199) return origin == "forgeton" with a pointer to the sibling forgeton-sdk's explain_error.

Compiled contract code

from kronos_sdk import KronosRegistry, RegistryInitConfig, load_registry_code

code = load_registry_code()  # bundled BoC → pytoniq Cell
registry = KronosRegistry.create_from_config(
    RegistryInitConfig(owner=owner_addr, treasury=treasury_addr),
    code,
    client=client,
)
await registry.send_deploy(wallet, value=int(0.5 * 10**9))

Surface map (TS ↔ Python)

TypeScript	Python
KronosRegistry.createFromAddress(addr)	KronosRegistry.create_from_address(addr, client=client)
KronosRegistry.createFromConfig(cfg, code)	KronosRegistry.create_from_config(cfg, code, client=client)
registry.sendRegisterJob(via, opts)	await registry.send_register_job(wallet, **opts)
registry.getJob(jobId)	await registry.get_job(job_id)
executionEconomics({...})	execution_economics(ExecutionEconomicsInput(...))
recommendedRegisterValue({...})	recommended_register_value(RecommendedRegisterValueInput(...))
jobWindowState({...})	job_window_state(JobWindowInput(...))
assignedAutomatonIndex({...})	assigned_automaton_index(AssignmentInput(...))
decodeEvent(body)	decode_event(body)
JobPresets.tight	JOB_PRESETS["tight"]
registerJobOpts(input, cfg)	register_job_opts(JobOptsInput(...), cfg)
previewJobCost(input, cfg)	preview_job_cost(PreviewJobCostInput(...), cfg)
validateRegisterOpts(opts, cfg)	validate_register_opts(...)
new KronosClient({ registry })	KronosClient(registry=registry)
client.jobs.balanceHealth(id)	await client.jobs.balance_health(id)
client.assignedAutomatonFor(id)	await client.assigned_automaton_for(id)
KRONOS_TESTNET.registry	KRONOS_TESTNET.registry

Sibling SDK — ForgeTON

The shared-security staking pool (ForgeTON) lives in a sibling Python SDK (forgeton-sdk). Install it alongside this package when you need pool-side operations: registering an automaton, increasing stake, watching slashes, decoding pool events.

Repository

Source lives in titon-network/kronos under sdks/python/. The TypeScript SDK and contract source live alongside it. File issues + PRs there.

License

MIT