Skip to main content

​
Client

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

client = TonapiWebhookClient("YOUR_API_KEY", Network.MAINNET)
ParameterDefaultDescription
api_keyRequiredAPI key from tonconsole.com.
networkRequiredMainnet or Testnet.
base_urlOptionalOverride the default URL.
timeout10Request timeout in seconds.
sessionOptionalExternal HTTP session to share across clients.
headersOptionalExtra HTTP headers.
cookiesOptionalExtra cookies.
retry_policyEnabledRetry policy, or None to disable.

​
Session

async with TonapiWebhookClient("YOUR_API_KEY", Network.MAINNET) as client:
    webhooks = await client.list()

​
Managing Webhooks

# Create a new webhook — returns a bound TonapiWebhook
webhook = await client.create("https://example.com/account-tx")
print(webhook.id, webhook.token)

# List all webhooks — returns list[WebhookInfo]
webhooks = await client.list()

# Find existing webhook by endpoint, or create a new one
webhook = await client.ensure("https://example.com/account-tx")

# Get a bound TonapiWebhook by ID
webhook = await client.get(webhook_id=42)

# Get full webhook info — raises TONAPINotFoundError if not found
info = await client.get_info(webhook_id=42)

# Delete a webhook and all its subscriptions
await client.delete(webhook_id=42)

​
Subscriptions

A bound TonapiWebhook manages subscriptions without requiring an explicit webhook ID.
PropertyTypeDescription
idintWebhook identifier.
endpointstrCallback URL.
tokenstrSecret token sent as Authorization: Bearer {token} — use for verifying incoming requests.

​
Account transactions

await webhook.subscribe(["EQ...", "EQ..."])
await webhook.unsubscribe(["EQ..."])
subs = await webhook.get_subscriptions(offset=0, limit=50)
await webhook.sync_accounts(["EQ...", "EQ..."])  # subscribe missing, unsubscribe extra

​
Other subscriptions

await webhook.subscribe_mempool_msg()
await webhook.unsubscribe_mempool_msg()

await webhook.subscribe_new_contracts()
await webhook.unsubscribe_new_contracts()

await webhook.subscribe_opcode_msg("0x7362d09c")
await webhook.unsubscribe_opcode_msg("0x7362d09c")

​
Deleting a webhook

await webhook.delete()

​
Event Handling

The dispatcher manages handler registration, webhook lifecycle, and event routing. 
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"],
)
ParameterDefaultDescription
url""Public webhook URL prefix.
clientOptionalTonapiWebhookClient (required for setup/teardown).
accountsOptionalAccount IDs to subscribe.
opcodesOptionalOpcodes for opcode_msg subscriptions.
**kwargs—Default dependencies injected into handler calls.

​
Registering handlers

@dispatcher.account_tx()
async def handler(event: AccountTxEvent) -> None: ...

@dispatcher.account_tx("EQ...", "EQ...", path="/custom")
async def handler(event: AccountTxEvent) -> None: ...

@dispatcher.mempool_msg()
async def handler(event: MempoolMsgEvent) -> None: ...

@dispatcher.opcode_msg()
async def handler(event: OpcodeMsgEvent) -> None: ...

@dispatcher.new_contracts()
async def handler(event: NewContractsEvent) -> None: ...
Decorators account_tx, opcode_msg, new_contracts accept *accounts for local filtering. All accept path for custom URL paths. Default paths: /account-tx, /mempool-msg, /opcode-msg, /new-contracts. Without decorators: 
from pytonapi.webhook import WebhookEventType

dispatcher.register(WebhookEventType.ACCOUNT_TX, handler, "EQ...")

​
Lifecycle

setup() opens a session, creates webhooks per event type, subscribes, and stores tokens. teardown(cleanup) closes the session. When cleanup=True, unsubscribes from all events first. Subscriptions persist across restarts — on next setup(), existing webhooks are reused via ensure(). 
await dispatcher.setup()
await dispatcher.teardown(cleanup=True)
Registered paths (only event types with handlers): 
dispatcher.paths  # {WebhookEventType.ACCOUNT_TX: "/account-tx", ...}

​
Processing events

await dispatcher.process(
    path="/account-tx",
    data=payload,
    authorization=headers.get("Authorization"),
)
The method validates the token, parses the event with Pydantic, filters by account if specified, and injects dependencies.

​
Dependencies

Pass default dependencies via **kwargs in the constructor. Handlers receive only parameters they declare in their signature: 
dispatcher = TonapiWebhookDispatcher(url="...", client=client, db=database)


@dispatcher.account_tx()
async def handler(event: AccountTxEvent, db: Database) -> None:
    await db.save(event)
Per-request dependencies passed to process() take priority over constructor defaults. Handlers can be async or sync — the dispatcher awaits async results automatically.

​
Models

FieldTypeDescription
idintWebhook ID.
endpointstrCallback URL.
tokenstrSecret token for Authorization header.
subscribed_accountsintNumber of subscribed accounts.
subscribed_msg_opcodesintNumber of subscribed opcodes.
subscribed_to_mempoolboolMempool subscription active.
subscribed_to_new_contractsboolNew contracts subscription active.
statusstrCurrent webhook status.
status_updated_atstrLast status update timestamp.
last_online_atstrLast online timestamp.
status_failed_attemptsintConsecutive failed delivery attempts.
FieldTypeDescription
account_idstrSubscribed account address.
last_delivered_ltintLast delivered logical time.
failed_atstr | NoneTimestamp of last failure.
failed_ltint | NoneLogical time of failed delivery.
failed_attemptsint | NoneNumber of failed delivery attempts.