Pytonapi (nessshon/tonapi)

PyTONAPI

https://github.com/nessshon/tonapi
Admin
Pytonapi is a Python SDK for accessing TON blockchain data via REST API, real-time streaming, and...

Tokens:61,904
Snippets:469
Trust Score:9.1
Update:1 day ago
Show doc for...
Context Summary (auto-generated)
Raw
# Pytonapi

Pytonapi is a Python SDK for accessing TON blockchain data via TONAPI services. It provides three main interfaces: a REST API client for querying accounts, NFTs, jettons, DNS records, and blockchain data; real-time streaming via Server-Sent Events (SSE) and WebSocket connections for monitoring transactions, blocks, and mempool messages; and a webhook system for receiving push notifications about on-chain events. The SDK is fully async, built on aiohttp, and uses Pydantic models for type-safe responses.

The library supports mainnet, testnet, and tetra networks with automatic rate limiting, retry policies with exponential backoff, and API key rotation for high-throughput applications. An API key is optional for REST endpoints (throttled to ~0.24 RPS without one) but required for streaming and webhooks. Keys can be obtained from tonconsole.com.

## REST API Client

### Initialize REST Client

The `TonapiRestClient` provides access to all REST API resources. Use it as an async context manager for automatic session management.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    # Basic initialization with API key and network
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        account = await tonapi.accounts.get_account(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA"
        )
        print(f"Balance: {account.balance}")

    # Without API key (rate-limited to 1 request per 4 seconds)
    async with TonapiRestClient(network=Network.MAINNET) as tonapi:
        account = await tonapi.accounts.get_account(account_id="donate.ton")
        print(f"Status: {account.status}")

asyncio.run(main())
```

### Get Account Information

Retrieve human-friendly account details including balance, status, interfaces, and name.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import raw_to_userfriendly, to_amount

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        account = await tonapi.accounts.get_account(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA"
        )

        # Convert raw address to user-friendly format
        print(f"Address: {raw_to_userfriendly(account.address)}")

        # Convert nanotons to TON (1 TON = 10^9 nanotons)
        print(f"Balance: {to_amount(account.balance)} TON")

        # Account status: active, uninit, frozen, nonexist
        print(f"Status: {account.status}")

        # Human-readable name (None if not recognized)
        print(f"Name: {account.name}")

asyncio.run(main())
```

### Get Account Transactions

Fetch blockchain transactions for an account with decoded message payloads.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import to_amount

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        result = await tonapi.blockchain.get_account_transactions(
            account_id="0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b",
            limit=100,
            sort_order="desc"  # "desc" (newest first) or "asc" (oldest first)
        )

        for tx in result.transactions:
            print(f"Value: {to_amount(tx.in_msg.value, precision=5)} TON")

            # Decode message payload (e.g., text comments)
            if tx.in_msg.decoded_op_name == "text_comment":
                print(f"Comment: {tx.in_msg.decoded_body['text']}")

asyncio.run(main())
```

### Get Account Events

Retrieve high-level account events with parsed actions like jetton transfers and NFT purchases.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        result = await tonapi.accounts.get_account_events(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA",
            limit=50,
            initiator=False,      # True = only events initiated by this account
            subject_only=False,   # True = filter where account is direct subject
            sort_order="desc"
        )

        for event in result.events:
            print(f"Event ID: {event.event_id}")
            print(f"Timestamp: {event.timestamp}")

            for action in event.actions:
                print(f"  Action: {action.type}")
                if action.simple_preview:
                    print(f"  Preview: {action.simple_preview.description}")

asyncio.run(main())
```

### Get Account Jetton Balances

Retrieve all jetton (token) balances for an account.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import to_amount

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        result = await tonapi.accounts.get_account_jettons_balances(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA",
            currencies=["usd", "ton"],  # Include fiat price conversion
            limit=100
        )

        for jetton in result.balances:
            decimals = jetton.jetton.decimals
            balance = to_amount(int(jetton.balance), decimals)
            print(f"{jetton.jetton.symbol}: {balance}")

            if jetton.price:
                print(f"  USD value: ${jetton.price.prices.get('USD', 0)}")

asyncio.run(main())
```

### Get Account NFTs

Retrieve all NFT items owned by an account.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import raw_to_userfriendly

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        result = await tonapi.accounts.get_account_nft_items(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA",
            limit=10,
            indirect_ownership=False  # True = include NFTs on sale
        )

        for nft in result.nft_items:
            address = raw_to_userfriendly(nft.address, is_bounceable=True)
            print(f"NFT: {address}")

            if nft.collection:
                col_addr = raw_to_userfriendly(nft.collection["address"], True)
                print(f"  Collection: {col_addr}")

            # Preview images: 5x5, 100x100, 500x500, 1500x1500
            if nft.previews:
                print(f"  Preview: {nft.previews[0].url}")

asyncio.run(main())
```

### Get Jetton Information

Retrieve metadata and statistics for a jetton (fungible token).

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # USDT jetton master address
        jetton_address = "EQCxE6mUtQJKFnGfaROTKOt1lZbDiiX1kCixRv7Nw2Id_sDs"

        jetton_info = await tonapi.jettons.get_jetton_info(
            account_id=jetton_address
        )

        print(f"Name: {jetton_info.metadata.name}")
        print(f"Symbol: {jetton_info.metadata.symbol}")
        print(f"Decimals: {jetton_info.metadata.decimals}")
        print(f"Total Supply: {jetton_info.total_supply}")
        print(f"Holders: {jetton_info.holders_count}")

asyncio.run(main())
```

### Get Jetton Holders

Retrieve the list of jetton holders with their balances.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import to_amount, raw_to_userfriendly

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # USDT jetton master address
        jetton_address = "EQCxE6mUtQJKFnGfaROTKOt1lZbDiiX1kCixRv7Nw2Id_sDs"

        result = await tonapi.jettons.get_jetton_holders(
            account_id=jetton_address,
            limit=100,
            offset=0
        )

        for holder in result.addresses:
            owner = raw_to_userfriendly(holder.owner.address)
            balance = to_amount(int(holder.balance), 6)  # USDT has 6 decimals
            print(f"{owner}: {balance}")

asyncio.run(main())
```

### Get NFT Collection

Retrieve NFT collection metadata and items.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import raw_to_userfriendly

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        collection_address = "EQAOQdwdw8kGftJCSFgOErM1mBjYPe4DBPq8-AhF6vr9si5N"

        # Get collection info
        collection = await tonapi.nft.get_collection(
            account_id=collection_address
        )
        print(f"Collection: {collection.metadata.get('name', 'Unknown')}")
        print(f"Items: {collection.next_item_index}")

        # Get items from collection
        items = await tonapi.nft.get_items_from_collection(
            account_id=collection_address,
            limit=10
        )

        for nft in items.nft_items:
            addr = raw_to_userfriendly(nft.address, True)
            print(f"  Item: {addr}")

asyncio.run(main())
```

### DNS Resolution

Resolve .ton and .t.me domain names to addresses.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.utils import raw_to_userfriendly

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Get domain info
        domain_info = await tonapi.dns.get_info(domain_name="foundation.ton")
        print(f"Domain: {domain_info.name}")
        print(f"Expiring: {domain_info.expiring_at}")

        # Resolve domain to wallet address
        dns_record = await tonapi.dns.resolve(domain_name="foundation.ton")
        if dns_record.wallet:
            wallet = raw_to_userfriendly(dns_record.wallet.address)
            print(f"Wallet: {wallet}")

asyncio.run(main())
```

### Get Token Rates

Retrieve token prices in various currencies (for display only, not financial transactions).

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Get rates for TON and USDT
        rates = await tonapi.rates.get_rates(
            tokens=["ton", "EQCxE6mUtQJKFnGfaROTKOt1lZbDiiX1kCixRv7Nw2Id_sDs"],
            currencies=["usd", "eur", "rub"]
        )

        for token, data in rates.get("rates", {}).items():
            print(f"{token}:")
            for currency, price in data.get("prices", {}).items():
                print(f"  {currency}: {price}")

asyncio.run(main())
```

### Get Blockchain Block

Retrieve blockchain block data and transactions.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Get latest masterchain head
        head = await tonapi.blockchain.get_masterchain_head()
        print(f"Latest seqno: {head.seqno}")
        print(f"Workchain: {head.workchain_id}")

        # Get block transactions
        transactions = await tonapi.blockchain.get_block_transactions(
            block_id=f"({head.workchain_id},{head.shard},{head.seqno})"
        )
        print(f"Transactions: {len(transactions.transactions)}")

asyncio.run(main())
```

### Execute Smart Contract Method

Execute GET methods on smart contracts.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Execute 'get_wallet_data' on a wallet contract
        result = await tonapi.blockchain.execute_get_method(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA",
            method_name="get_wallet_data",
            args=[]  # Method arguments as strings
        )

        print(f"Success: {result.success}")
        print(f"Exit code: {result.exit_code}")

        for item in result.stack:
            print(f"  {item.type}: {item.value}")

asyncio.run(main())
```

### Gasless Transactions

Get gasless transfer configuration. Gas fees are paid in supported jettons instead of TON.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Get supported jettons for gasless transfers
        config = await tonapi.gasless.config()

        print("Relay address:", config.relay_address)
        print("Supported jettons:")
        for jetton in config.gas_jettons:
            print(f"  {jetton.master_id}")

asyncio.run(main())
```

### Staking Pools

Retrieve staking pool information and account participation.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        # Get all available staking pools
        pools = await tonapi.staking.get_pools(include_unverified=False)

        for pool in pools.get("implementations", {}).values():
            for p in pool.get("pools", []):
                print(f"Pool: {p.get('name', 'Unknown')}")
                print(f"  APY: {p.get('apy', 0):.2f}%")

        # Get account's staking participation
        account_staking = await tonapi.staking.get_account_nominators_pools(
            account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA"
        )

        for pool in account_staking.pools:
            print(f"Staked in: {pool.pool}")

asyncio.run(main())
```

### API Key Rotation

Configure multiple API keys for automatic rotation on rate limit errors.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network, ApiKey

async def main():
    # Configure multiple keys with per-key rate limits
    keys = [
        ApiKey("KEY_1", rps_limit=10, rps_period=1.0),
        ApiKey("KEY_2", rps_limit=10, rps_period=1.0),
        ApiKey("KEY_3", rps_limit=10, rps_period=1.0),
    ]

    # Client automatically rotates to next key on HTTP 429
    async with TonapiRestClient(keys, Network.MAINNET) as tonapi:
        for i in range(100):
            account = await tonapi.accounts.get_account(
                account_id="EQAUxYSo-UwoqAGixaD3d7CNLp9PthgmEZfnr6BvsijzJHdA"
            )
            print(f"Request {i}: {account.balance}")

asyncio.run(main())
```

## Streaming API

### SSE (Server-Sent Events) Subscriptions

Subscribe to real-time blockchain events using SSE transport.

```python
import asyncio
from pytonapi.streaming import TonapiStreaming
from pytonapi.types import Network, Opcode, Workchain

API_KEY = "YOUR_API_KEY"
ACCOUNT_ID = "0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b"

def make_stop(seconds: float) -> asyncio.Event:
    """Create a stop event that fires after specified seconds."""
    event = asyncio.Event()

    async def _set_after():
        await asyncio.sleep(seconds)
        event.set()

    asyncio.ensure_future(_set_after())
    return event

async def main():
    async with TonapiStreaming(API_KEY, Network.MAINNET) as streaming:
        # Subscribe to account transactions
        # accounts=None subscribes to ALL accounts (high volume!)
        print("Subscribing to transactions...")
        async for event in streaming.sse.subscribe_transactions(
            accounts=[ACCOUNT_ID],
            operations=[Opcode.TEXT_COMMENT, Opcode.JETTON_TRANSFER],
            stop=make_stop(30)
        ):
            print(f"TX: {event.account_id} | {event.tx_hash}")

        # Subscribe to new blocks
        print("Subscribing to blocks...")
        async for block in streaming.sse.subscribe_blocks(
            workchain=Workchain.MASTERCHAIN,
            stop=make_stop(30)
        ):
            print(f"Block: seqno={block.seqno}, workchain={block.workchain}")

        # Subscribe to completed traces
        print("Subscribing to traces...")
        async for trace in streaming.sse.subscribe_traces(
            accounts=[ACCOUNT_ID],
            stop=make_stop(30)
        ):
            print(f"Trace: {trace.hash}")

        # Subscribe to mempool (pending messages)
        print("Subscribing to mempool...")
        async for msg in streaming.sse.subscribe_mempool(
            stop=make_stop(30)
        ):
            print(f"Mempool: {msg.boc[:50]}...")

asyncio.run(main())
```

### WebSocket Subscriptions

Subscribe to real-time events using WebSocket transport with JSON-RPC protocol.

```python
import asyncio
from pytonapi.streaming import TonapiStreaming
from pytonapi.types import Network, Opcode, Workchain

API_KEY = "YOUR_API_KEY"
ACCOUNT_ID = "0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b"

def make_stop(seconds: float) -> asyncio.Event:
    event = asyncio.Event()
    asyncio.get_event_loop().call_later(seconds, event.set)
    return event

async def main():
    async with TonapiStreaming(API_KEY, Network.MAINNET) as streaming:
        # WebSocket transactions subscription
        print("WebSocket: Subscribing to transactions...")
        async for event in streaming.ws.subscribe_transactions(
            accounts=[ACCOUNT_ID],
            operations=[Opcode.JETTON_TRANSFER],
            stop=make_stop(60)
        ):
            print(f"WS TX: {event.account_id} | {event.tx_hash}")

        # WebSocket blocks subscription
        print("WebSocket: Subscribing to blocks...")
        async for block in streaming.ws.subscribe_blocks(
            workchain=Workchain.MASTERCHAIN,
            stop=make_stop(60)
        ):
            print(f"WS Block: seqno={block.seqno}")

        # WebSocket traces subscription
        print("WebSocket: Subscribing to traces...")
        async for trace in streaming.ws.subscribe_traces(
            accounts=None,  # All traces
            stop=make_stop(60)
        ):
            print(f"WS Trace: {trace.hash}")

asyncio.run(main())
```

## Webhook API

### Webhook Client

Create and manage webhooks for push notifications.

```python
import asyncio
from pytonapi.webhook import TonapiWebhookClient
from pytonapi.types import Network

API_KEY = "YOUR_API_KEY"
ACCOUNT_ID = "0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b"

async def main():
    async with TonapiWebhookClient(API_KEY, Network.MAINNET) as client:
        # Create a new webhook
        webhook = await client.create(endpoint="https://example.com/webhook/account-tx")
        print(f"Webhook ID: {webhook.id}")
        print(f"Token: {webhook.token}")  # Use for Authorization header verification

        # Subscribe to account transactions
        await webhook.subscribe(accounts=[ACCOUNT_ID])

        # Get current subscriptions
        subs = await webhook.get_subscriptions(limit=100)
        for sub in subs:
            print(f"Subscribed: {sub.account_id}")

        # Sync accounts (add missing, remove extra)
        await webhook.sync_accounts([ACCOUNT_ID, "another_account"])

        # Subscribe to other event types
        await webhook.subscribe_mempool_msg()
        await webhook.subscribe_new_contracts()
        await webhook.subscribe_opcode_msg(opcode="0x0f8a7ea5")  # Jetton transfer

        # List all webhooks
        webhooks = await client.list()
        for w in webhooks:
            print(f"Webhook {w.id}: {w.endpoint}")

        # Get or create webhook (reuses existing if endpoint matches)
        webhook = await client.ensure(endpoint="https://example.com/webhook/account-tx")

        # Delete webhook
        await webhook.delete()

asyncio.run(main())
```

### Webhook Dispatcher with FastAPI

Handle incoming webhook notifications with automatic routing and token verification.

```python
from contextlib import asynccontextmanager
import uvicorn
from fastapi import FastAPI, Request, Response

from pytonapi.types import Network, Opcode
from pytonapi.webhook import (
    AccountTxEvent,
    MempoolMsgEvent,
    NewContractsEvent,
    OpcodeMsgEvent,
    TonapiWebhookClient,
    TonapiWebhookDispatcher,
)

API_KEY = "YOUR_API_KEY"
WEBHOOK_URL = "https://example.com/webhook"
ACCOUNT_ID = "0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b"

# Create client and dispatcher
client = TonapiWebhookClient(API_KEY, Network.MAINNET)
dispatcher = TonapiWebhookDispatcher(
    WEBHOOK_URL,
    client=client,
    accounts=[ACCOUNT_ID],
    opcodes=[Opcode.TEXT_COMMENT],
)

# Register event handlers
@dispatcher.account_tx()
async def on_account_tx(event: AccountTxEvent) -> None:
    """Handle all account transaction events."""
    print(f"Account TX: {event.account_id} | {event.tx_hash}")

@dispatcher.account_tx(ACCOUNT_ID)
async def on_my_account_tx(event: AccountTxEvent) -> None:
    """Handle transactions for specific account only."""
    print(f"My TX: {event.tx_hash}")

@dispatcher.mempool_msg()
async def on_mempool_msg(event: MempoolMsgEvent) -> None:
    """Handle mempool (pending) messages."""
    print(f"Mempool: {event.boc[:50]}...")

@dispatcher.opcode_msg()
async def on_opcode_msg(event: OpcodeMsgEvent) -> None:
    """Handle messages with specific opcodes."""
    print(f"Opcode: {event.account_id} | {event.tx_hash}")

@dispatcher.new_contracts()
async def on_new_contract(event: NewContractsEvent) -> None:
    """Handle new contract deployments."""
    print(f"New contract: {event.account_id}")

# HTTP handler for webhook POST requests
async def handle_webhook(request: Request) -> Response:
    data = await request.json()
    try:
        authorization = request.headers.get("Authorization")
        await dispatcher.process(request.url.path, data, authorization=authorization)
    except Exception as e:
        print(f"Webhook error: {e}")
        return Response(status_code=401)
    return Response(status_code=200)

# FastAPI lifespan for setup/teardown
@asynccontextmanager
async def lifespan(application: FastAPI):
    await dispatcher.setup()  # Creates webhooks, subscribes, stores tokens

    # Register routes for each event type
    for path in dispatcher.paths.values():
        application.add_api_route(path, handle_webhook, methods=["POST"])

    yield

    await dispatcher.teardown()  # Use teardown(cleanup=True) to unsubscribe

app = FastAPI(lifespan=lifespan)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

## Utility Functions

### Address Conversion

Convert between raw and user-friendly TON address formats.

```python
from pytonapi.utils import raw_to_userfriendly, userfriendly_to_raw

# Raw address format: workchain:hex_hash
raw_address = "0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b"

# Convert to user-friendly (base64)
user_friendly = raw_to_userfriendly(
    raw_address,
    is_bounceable=True,   # True for smart contracts, False for wallets
    is_url_safe=True,     # URL-safe base64 encoding
    is_test_only=False    # True for testnet addresses
)
print(f"User-friendly: {user_friendly}")
# Output: EQBAjaOyi2wGWlk-EDkSabuqnF-MrrwMadnwqrurKpklawBA

# Convert back to raw format
raw_again = userfriendly_to_raw(user_friendly)
print(f"Raw: {raw_again}")
# Output: 0:408da3b28b6c065a593e10391269baaa9c5f8caebc0c69d9f0aabbab2a99256b
```

### Amount Conversion

Convert between human-readable amounts and nanotons (smallest units).

```python
from pytonapi.utils import to_amount, to_nano

# TON uses 9 decimal places (1 TON = 10^9 nanotons)
nanotons = 1_500_000_000  # 1.5 TON in nanotons

# Convert to human-readable
amount = to_amount(nanotons)
print(f"Amount: {amount} TON")  # Output: 1.5 TON

# With precision (rounds down)
amount_precise = to_amount(nanotons, precision=2)
print(f"Precise: {amount_precise} TON")  # Output: 1.50 TON

# Convert human-readable to nanotons
nano = to_nano(1.5)
print(f"Nanotons: {nano}")  # Output: 1500000000

# For jettons with different decimals (e.g., USDT has 6)
usdt_nano = to_nano(100.50, decimals=6)
print(f"USDT units: {usdt_nano}")  # Output: 100500000

usdt_amount = to_amount(100500000, decimals=6)
print(f"USDT amount: {usdt_amount}")  # Output: 100.5
```

## Error Handling

Handle TONAPI-specific errors with detailed status information.

```python
import asyncio
from pytonapi.rest import TonapiRestClient
from pytonapi.types import Network
from pytonapi.exceptions import (
    TONAPIError,
    TONAPIBadRequestError,
    TONAPIUnauthorizedError,
    TONAPINotFoundError,
    TONAPITooManyRequestsError,
    TONAPIConnectionError,
    TONAPIValidationError,
)

async def main():
    async with TonapiRestClient("YOUR_API_KEY", Network.MAINNET) as tonapi:
        try:
            account = await tonapi.accounts.get_account(
                account_id="invalid_address"
            )
        except TONAPIBadRequestError as e:
            print(f"Bad request (400): {e.message}")
            print(f"Hint: {e.hint}")
        except TONAPIUnauthorizedError as e:
            print(f"Unauthorized (401): {e.message}")
        except TONAPINotFoundError as e:
            print(f"Not found (404): {e.message}")
        except TONAPITooManyRequestsError as e:
            print(f"Rate limited (429): {e.message}")
        except TONAPIConnectionError as e:
            print(f"Connection error: {e.message}")
        except TONAPIValidationError as e:
            print(f"Validation error: {e.message}")
            print(f"Model: {e.model.__name__}")
        except TONAPIError as e:
            print(f"General error: {e.message}")

asyncio.run(main())
```

## Types and Constants

Common types and constants used throughout the SDK.

```python
from pytonapi.types import (
    Network,          # MAINNET, TESTNET, TETRA
    Workchain,        # MASTERCHAIN (-1), BASECHAIN (0)
    Opcode,           # Well-known message opcodes
    ApiKey,           # API key with rate limiting config
    RetryPolicy,      # Retry configuration for HTTP errors
    ReconnectPolicy,  # Reconnect configuration for streaming
)

# Networks
print(Network.MAINNET)   # -239
print(Network.TESTNET)   # -3

# Workchains
print(Workchain.MASTERCHAIN)  # -1
print(Workchain.BASECHAIN)    # 0

# Common opcodes for filtering
print(Opcode.TEXT_COMMENT)       # 0x00000000
print(Opcode.JETTON_TRANSFER)    # 0x0f8a7ea5
print(Opcode.NFT_TRANSFER)       # 0x5fcc3d14
print(Opcode.JETTON_BURN)        # 0x595f07bc

# Custom API key with rate limiting
key = ApiKey(
    key="YOUR_KEY",
    rps_limit=10,    # Max requests per second
    rps_period=1.0   # Window size in seconds
)
```

## Summary

Pytonapi provides comprehensive access to the TON blockchain through TONAPI services. The REST client covers all major blockchain operations including account queries, NFT and jetton management, DNS resolution, staking pools, and smart contract execution. The streaming API enables real-time monitoring of transactions, blocks, traces, and mempool messages through SSE or WebSocket connections with automatic reconnection. The webhook system offers push-based notifications with a dispatcher pattern that handles routing, token verification, and automatic subscription management.

The SDK is designed for production use with built-in rate limiting, retry policies with exponential backoff, API key rotation for high-throughput scenarios, and comprehensive error handling. All responses are parsed into Pydantic models for type safety, and utility functions simplify common tasks like address conversion and amount formatting. Integration is straightforward: install with `pip install pytonapi`, obtain an API key from tonconsole.com, and use async context managers for automatic resource cleanup.