TONAPI Python SDK

The pytonapi streaming module allows you to subscribe to real-time blockchain events using either Server-Sent Events (SSE) or WebSockets. Both transports multiplex all subscriptions over a single connection and use a decorator-based API to handle incoming events.

Transport Options

• TonapiSSE: A one-way stream. It uses a single POST request where the server pushes events to the client.
• TonapiWebSocket: A bidirectional stream. It allows for dynamic subscribing and unsubscribing after the initial connection is established.

Note: An API key is required for all streaming connections. Unauthenticated connections will be rejected. You can obtain a key at tonconsole.com.

from pytonapi.streaming import TonapiSSE, TonapiWebSocket

# For SSE
sse = TonapiSSE("YOUR_API_KEY", Network.MAINNET)

# For WebSocket
ws = TonapiWebSocket("YOUR_API_KEY", Network.MAINNET)

Webhooks deliver real-time POST notifications to your server when blockchain events occur. The SDK provides three main components to manage this:

1. TonapiWebhookClient: A low-level HTTP client used for CRUD operations on webhooks (create, list, delete, subscribe/unsubscribe).
2. TonapiWebhook: A bound webhook object that manages subscriptions without requiring you to handle explicit IDs manually.
3. TonapiWebhookDispatcher: A high-level, decorator-based dispatcher used to register event handlers, automatically set up webhook subscriptions, and process incoming HTTP requests.

Security Note: Every webhook has a unique secret token. TON API sends this token in the Authorization: Bearer {token} header with every event. The TonapiWebhookDispatcher.process() method verifies this token automatically to ensure requests are genuine.

from pytonapi.webhook import TonapiWebhookClient, TonapiWebhookDispatcher

client = TonapiWebhookClient("YOUR_API_KEY", Network.MAINNET)
dispatcher = TonapiWebhookDispatcher(
    "https://example.com/webhook",
    client=client,
    accounts=["ACCOUNT_ADDRESS"],
)

The client manages an underlying HTTP session. You can manage its lifecycle using a context manager, manual calls, or by providing an external aiohttp.ClientSession.

# Context manager (Recommended)
async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
    account = await tonapi.accounts.get_account(account_id="EQ...")

# Manual lifecycle management
tonapi = TonapiRestClient("YOUR_API_KEY", Network.MAINNET)
await tonapi.create_session()
try:
    account = await tonapi.accounts.get_account(account_id="EQ...")
finally:
    await tonapi.close_session()

# Using an external session
import aiohttp

async with aiohttp.ClientSession() as session:
    async with TonapiRestClient(
        "YOUR_API_KEY",
        Network.MAINNET,
        session=session,
    ) as tonapi:
        account = await tonapi.accounts.get_account(account_id="EQ...")
    # Note: The client does not close the external session; you must manage it.

The TonapiWebhookDispatcher manages handler registration, webhook lifecycle, and event routing. It is designed to receive incoming HTTP requests and route them to the correct handler.

Initialization:

dispatcher = TonapiWebhookDispatcher(
    url="https://example.com",
    client=client,
    accounts=["EQ..."],
    opcodes=["0x7362d09c"],
    db=database # Example dependency
)

Registering Handlers: Use decorators to register handlers. Decorators can accept *accounts for local filtering and a path for custom URL paths. Default paths are /account-tx, /mempool-msg, /opcode-msg, and /new-contracts.

• @dispatcher.account_tx()
• @dispatcher.mempool_msg()
• @dispatcher.opcode_msg()
• @dispatcher.new_contracts()

Lifecycle:

• await dispatcher.setup(): Opens a session, creates webhooks, and subscribes to events. Existing webhooks are reused via ensure().
• await dispatcher.teardown(cleanup=True): Closes the session. If cleanup=True, it unsubscribes from all events first.

Processing Events: When your web server receives a request, pass the data to process():

await dispatcher.process(
    path="/account-tx",
    data=payload,
    authorization=headers.get("Authorization"),
)

Dependencies: Dependencies passed to the constructor via **kwargs are injected into handlers. Per-request dependencies passed to process() take priority.

from pytonapi.webhook import TonapiWebhookClient, TonapiWebhookDispatcher
from pytonapi.types import Network

client = TonapiWebhookClient("YOUR_API_KEY", Network.MAINNET)

dispatcher = TonapiWebhookDispatcher(
    "https://example.com",
    client=client,
    accounts=["EQ..."],
    opcodes=["0x7362d09c"],
)

@dispatcher.account_tx()
async def handler(event: AccountTxEvent, db: Database) -> None:
    await db.save(event)

# Lifecycle management
await dispatcher.setup()
# ... receive requests ...
await dispatcher.teardown(cleanup=True)