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

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)

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