Overview

import asyncio

from toncenter.types import Network
from toncenter.streaming import Finality, ToncenterSSE, TransactionsNotification
from toncenter.exceptions import ToncenterConnectionLostError

sse = ToncenterSSE("YOUR_API_KEY", Network.MAINNET)


@sse.on_transactions(min_finality=Finality.FINALIZED)
async def on_tx(n: TransactionsNotification) -> None:
    for tx in n.transactions:
        print(tx.get("hash"))


async def main() -> None:
    try:
        await sse.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"])
    except ToncenterConnectionLostError as e:
        print(f"Connection lost after {e.attempts} reconnect attempts")
    finally:
        await sse.stop()


if __name__ == "__main__":
    asyncio.run(main())