Try Live
Add Docs
Rankings
Pricing
Docs
Install
Theme
Install
Docs
Pricing
More...
More...
Try Live
Rankings
Enterprise
Create API Key
Add Docs
toncenter
https://github.com/nessshon/toncenter
Admin
TON Center is a Python SDK for accessing TON blockchain data via REST API (v2 and v3) and real-time
...
Tokens:
124,247
Snippets:
632
Trust Score:
9.1
Update:
2 weeks ago
Context
Skills
Chat
Benchmark
93.8
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# TON Center TON Center is a Python SDK for accessing TON blockchain data via REST API and real-time streaming (SSE & WebSocket). It provides two complementary APIs: REST API v2 for direct liteserver queries (accounts, transactions, blocks, smart contracts) and REST API v3 for indexed and enriched data (Jettons, NFTs, DNS, actions, traces). The SDK requires Python 3.10+ with aiohttp and pydantic dependencies. The SDK supports both Mainnet and Testnet networks with optional API keys (required for streaming, optional for REST at ~1 RPS without key). It features automatic retries with exponential backoff for rate-limited and server error responses, session management via async context managers, and a decorator-based handler API for real-time streaming events with finality levels (pending, confirmed, finalized). ## Installation ```bash pip install toncenter ``` ## Quick Start ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: info = await client.v2.accounts.get_address_information( "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" ) print(info.balance) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Accounts ### Get Address Information Retrieve detailed account information including balance, state, code, and data for any TON address. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.accounts.get_address_information( ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Get Address Balance Get the balance of a TON address in nanotons. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.accounts.get_address_balance( ADDRESS, ) print(result) if __name__ == "__main__": asyncio.run(main()) ``` ### Get Wallet Information Retrieve wallet-specific information including wallet type, seqno, and public key. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "UQCDrgGaI6gWK-qlyw69xWZosurGxrpRgIgSkVsgahUtxZR0" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.accounts.get_wallet_information( ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Transactions ### Get Transactions Retrieve transaction history for a specific address with optional pagination using lt and hash parameters. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.transactions.get_transactions( ADDRESS, ) for tx in result: print(tx.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Blocks ### Get Masterchain Info Get the current masterchain block information including last block seqno and state root hash. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.blocks.get_masterchain_info() print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Send ### Send BOC Broadcast a serialized bag-of-cells (BOC) message to the TON network for transaction execution. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network BOC = "te6ccgEBAQEAAgAAAA==" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.send.send_boc( BOC, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Estimate Fee Calculate the estimated fee for a transaction before sending it to the network. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "UQCDrgGaI6gWK-qlyw69xWZosurGxrpRgIgSkVsgahUtxZR0" BODY = "te6ccgEBAQEAAgAAAA==" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.send.estimate_fee( ADDRESS, BODY, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Run Method ### Run Get Method Execute a smart contract get method with optional stack parameters and return the result. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "EQC3dNlesgVD8YbAazcauIrXBPfiVhMMr5YYk2in0Mtsz0Bz" METHOD = "dnsresolve" STACK = [ ["tvm.Slice", "te6ccgEBAQEABwAACm5lc3MA"], ["num", "0x19f02441ee588fdb26ee24b2568dd035c3c9206e11ab979be62e55558a1d17ff"], ] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.runmethod.run_get_method( ADDRESS, METHOD, STACK, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - Utils ### Detect Address Parse and validate a TON address, returning all address formats (raw, bounceable, non-bounceable). ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.utils.detect_address( ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V2 - RPC ### JSON RPC Execute any V2 API method via JSON-RPC interface for flexibility in method invocation. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network METHOD = "getAddressInformation" PARAMS = {"address": "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"} async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v2.rpc.json_rpc( METHOD, params=PARAMS, ) print(result) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - Accounts ### Get Account States Retrieve account states for multiple addresses in a single batch request from the indexed database. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = ["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.accounts.get_account_states( address=ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - Blockchain ### Get Transactions (V3) Query transactions from the indexed PostgreSQL database with advanced filtering options. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ACCOUNT = ["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.blockchain.get_transactions( account=ACCOUNT, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - Actions ### Get Actions Retrieve classified actions (transfers, swaps, etc.) for an account from the indexed database. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ACCOUNT = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.actions.get_actions( account=ACCOUNT, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Get Traces Retrieve full trace trees with transactions and actions for transaction analysis. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ACCOUNT = "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.actions.get_traces( account=ACCOUNT, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - Jettons ### Get Jetton Masters Retrieve Jetton (token) contract metadata including name, symbol, decimals, and total supply. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = ["EQCxE6mUtQJKFnGfaROTKOt1lZbDiiX1kCixRv7Nw2Id_sDs"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.jettons.get_jetton_masters( address=ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Get Jetton Wallets Retrieve Jetton wallet information including balance for a specific owner address. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network OWNER_ADDRESS = ["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.jettons.get_jetton_wallets( owner_address=OWNER_ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Get Jetton Transfers Query Jetton transfer history for a specific token contract. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network JETTON_MASTER = "EQCxE6mUtQJKFnGfaROTKOt1lZbDiiX1kCixRv7Nw2Id_sDs" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.jettons.get_jetton_transfers( jetton_master=JETTON_MASTER, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - NFTs ### Get NFT Collections Retrieve NFT collection metadata including name, description, and owner information. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network COLLECTION_ADDRESS = ["EQC3dNlesgVD8YbAazcauIrXBPfiVhMMr5YYk2in0Mtsz0Bz"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.nfts.get_nft_collections( collection_address=COLLECTION_ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` ### Get NFT Items Retrieve NFT item details including metadata, owner, and collection information. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network ADDRESS = ["EQAMrsze7MaG_7P2ENd1eeT-S2VttJ1myT9sX5f-F1gY7xGx"] async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.nfts.get_nft_items( address=ADDRESS, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## REST API V3 - DNS ### Get DNS Records Resolve TON DNS domain names to their associated records and wallet addresses. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network DOMAIN = "ness.ton" async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: result = await client.v3.dns.get_dns_records( domain=DOMAIN, ) print(result.model_dump_json(indent=4)) if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming API - SSE ### SSE Transaction Streaming Subscribe to real-time transaction events via Server-Sent Events with finality filtering. ```python import asyncio from toncenter.types import Network from toncenter.streaming import Finality, ToncenterSSE, TransactionsNotification 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"]) finally: await sse.stop() if __name__ == "__main__": asyncio.run(main()) ``` ### SSE Actions Streaming Subscribe to classified blockchain actions with type filtering (transfers, swaps, etc.). ```python import asyncio from toncenter.types import Network from toncenter.streaming import ( ActionsNotification, ActionType, Finality, ToncenterSSE, ) sse = ToncenterSSE("YOUR_API_KEY", Network.MAINNET) @sse.on_actions( min_finality=Finality.FINALIZED, action_types=[ActionType.JETTON_TRANSFER, ActionType.TON_TRANSFER], ) async def on_action(n: ActionsNotification) -> None: for a in n.actions: print(a.get("type")) async def main() -> None: try: await sse.start( addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"], include_address_book=True, include_metadata=True, ) finally: await sse.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming API - WebSocket ### WebSocket Transaction Streaming Subscribe to real-time transaction events via WebSocket with bidirectional communication. ```python import asyncio from toncenter.types import Network from toncenter.streaming import Finality, ToncenterWebSocket, TransactionsNotification ws = ToncenterWebSocket("YOUR_API_KEY", Network.MAINNET) @ws.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 ws.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"]) finally: await ws.stop() if __name__ == "__main__": asyncio.run(main()) ``` ### WebSocket Dynamic Subscription Modify subscriptions on the fly without reconnecting using WebSocket's dynamic subscription feature. ```python import asyncio from toncenter.types import Network from toncenter.streaming import Finality, ToncenterWebSocket, TransactionsNotification ws = ToncenterWebSocket("YOUR_API_KEY", Network.MAINNET) @ws.on_transactions(min_finality=Finality.CONFIRMED) async def on_tx(n: TransactionsNotification) -> None: for tx in n.transactions: print(tx.get("hash")) async def main() -> None: try: await ws.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"]) # Wait for connection await ws.wait_subscribed(timeout=10.0) # Dynamically change subscription await ws.dynamic_subscribe( addresses=["0:ed1691307050047117b998b561d8de82d31fbf84910ced6eb5fc92e7485ef8a7"], min_finality=Finality.CONFIRMED, include_metadata=True, ) # Later, unsubscribe from specific addresses await ws.dynamic_unsubscribe( addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"] ) finally: await ws.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming - Reconnect Policy ### Custom Reconnect Policy Configure automatic reconnection behavior with exponential backoff for transient failures. ```python import asyncio from toncenter.types import Network, ReconnectPolicy from toncenter.streaming import Finality, ToncenterWebSocket, TransactionsNotification policy = ReconnectPolicy( max_reconnects=10, delay=2.0, max_delay=30.0, backoff_factor=2.0, ) ws = ToncenterWebSocket("YOUR_API_KEY", Network.MAINNET, reconnect_policy=policy) @ws.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 ws.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"]) finally: await ws.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming - Connection State Monitoring ### Connection State Callback Monitor connection state transitions (IDLE, CONNECTING, SUBSCRIBED, RECONNECTING) via callbacks. ```python import asyncio from toncenter.types import Network from toncenter.streaming import ( ConnectionState, Finality, ToncenterWebSocket, TransactionsNotification, ) def on_state(state: ConnectionState) -> None: print(f"Connection state: {state.value}") ws = ToncenterWebSocket("YOUR_API_KEY", Network.MAINNET, on_state_change=on_state) @ws.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 ws.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"]) finally: await ws.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming - Account State and Jetton Events ### Account State and Jetton Change Handlers Subscribe to account state changes and Jetton wallet balance updates in real-time. ```python import asyncio from toncenter.types import Network from toncenter.streaming import ( AccountStateNotification, Finality, JettonsNotification, ToncenterSSE, ) sse = ToncenterSSE("YOUR_API_KEY", Network.MAINNET) @sse.on_account_states(min_finality=Finality.FINALIZED) async def on_account_state(n: AccountStateNotification) -> None: if n.state: print(f"Account: {n.account}, Balance: {n.state.balance}") @sse.on_jettons(min_finality=Finality.FINALIZED) async def on_jetton_change(n: JettonsNotification) -> None: if n.jetton: print(f"Jetton wallet: {n.jetton.address}, Balance: {n.jetton.balance}") async def main() -> None: try: await sse.start( addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"], include_metadata=True, ) finally: await sse.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Streaming - Trace Invalidation ### Trace Invalidation Handler Handle trace invalidation events when pending or confirmed traces are rolled back. ```python import asyncio from toncenter.types import Network from toncenter.streaming import ( Finality, ToncenterSSE, TraceInvalidatedNotification, TransactionsNotification, ) sse = ToncenterSSE("YOUR_API_KEY", Network.MAINNET) @sse.on_transactions(min_finality=Finality.PENDING) async def on_tx(n: TransactionsNotification) -> None: for tx in n.transactions: print(f"Transaction: {tx.get('hash')}") @sse.on_trace_invalidated async def on_invalidated(n: TraceInvalidatedNotification) -> None: print(f"Trace invalidated: {n.trace_external_hash_norm}") # Discard cached data for this trace async def main() -> None: try: await sse.start(addresses=["EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2"]) finally: await sse.stop() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Client Configuration ### Custom Session and Retry Policy Configure custom aiohttp sessions, retry policies, and rate limiting for advanced use cases. ```python import asyncio import aiohttp from toncenter.rest import ToncenterRestClient from toncenter.types import Network, RetryPolicy, RetryRule async def main() -> None: # Custom retry policy retry_policy = RetryPolicy(rules=( RetryRule( statuses=frozenset({429}), max_retries=10, base_delay=0.5, max_delay=10.0, backoff_factor=2.0, ), )) # Using external session async with aiohttp.ClientSession() as session: async with ToncenterRestClient( network=Network.MAINNET, session=session, retry_policy=retry_policy, rps_limit=5, # Client-side rate limit timeout=30, ) as client: balance = await client.v2.accounts.get_address_balance( "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" ) print(balance) if __name__ == "__main__": asyncio.run(main()) ``` --- ## Utility Functions ### Address and Amount Conversion Convert between address formats and amount units using built-in utility functions. ```python from toncenter.utils import raw_to_userfriendly, userfriendly_to_raw, to_nano, to_amount # Address conversion raw = "0:83ae019a23a8162beaa5cb0ebdc56668b2eac6c6ba51808812915b206a152dc5" # Raw to user-friendly (non-bounceable by default) print(raw_to_userfriendly(raw)) # UQCDrgGaI6gWK-qlyw69xWZosurGxrpRgIgSkVsgahUtxZR0 # Raw to user-friendly (bounceable) print(raw_to_userfriendly(raw, is_bounceable=True)) # EQCDrgGaI6gWK-qlyw69xWZosurGxrpRgIgSkVsgahUtxcmx # User-friendly to raw print(userfriendly_to_raw("EQCDrgGaI6gWK-qlyw69xWZosurGxrpRgIgSkVsgahUtxcmx")) # 0:83ae019a23a8162beaa5cb0ebdc56668b2eac6c6ba51808812915b206a152dc5 # Amount conversion (TON uses 9 decimals) print(to_nano(1.5)) # 1_500_000_000 print(to_nano("0.001")) # 1_000_000 print(to_nano(100, decimals=6)) # 100_000_000 (USDT) print(to_amount(1_500_000_000)) # Decimal("1.5") print(to_amount(1_500_000_000, precision=2)) # Decimal("1.50") print(to_amount(100_000_000, decimals=6)) # Decimal("100") ``` --- ## Error Handling ### Exception Handling Handle SDK exceptions with the comprehensive error hierarchy for robust applications. ```python import asyncio from toncenter.rest import ToncenterRestClient from toncenter.types import Network from toncenter.exceptions import ( ToncenterError, ToncenterBadRequestError, ToncenterNotFoundError, ToncenterTooManyRequestsError, ToncenterRetryError, ) async def main() -> None: async with ToncenterRestClient(network=Network.MAINNET) as client: try: result = await client.v2.accounts.get_address_information( "EQDtFpEwcFAEcRe5mLVh2N6C0x-_hJEM7W61_JLnSF74p4q2" ) print(result.balance) except ToncenterBadRequestError as e: print(f"Bad request: {e}") except ToncenterNotFoundError as e: print(f"Resource not found: {e}") except ToncenterTooManyRequestsError as e: print(f"Rate limited: {e}") except ToncenterRetryError as e: print(f"All retry attempts exhausted: {e}") except ToncenterError as e: print(f"SDK error: {e}") if __name__ == "__main__": asyncio.run(main()) ``` --- ## Summary TON Center SDK is designed for developers building applications on the TON blockchain who need reliable access to blockchain data. The primary use cases include: wallet applications requiring real-time balance updates and transaction monitoring, DeFi platforms needing Jetton transfer tracking and swap detection, NFT marketplaces querying collections and items, analytics dashboards aggregating blockchain activity, and bots requiring instant notification of specific address activity. The SDK's dual API approach (V2 for real-time liteserver queries, V3 for indexed historical data) allows applications to choose the appropriate data source based on latency vs. richness requirements. Integration patterns typically involve using the REST client with async context managers for one-off queries, combining V2 and V3 APIs to get both real-time state and historical context, and leveraging streaming for event-driven architectures. The decorator-based handler API simplifies subscription management, while the finality system (pending/confirmed/finalized) enables applications to balance between speed and certainty. For production deployments, configure API keys for higher rate limits, implement proper error handling with the exception hierarchy, use custom retry and reconnect policies, and consider external aiohttp sessions for connection pooling across multiple SDK instances.