import aiohttp
from nacl.exceptions import BadSignatureError
from ton_core import Address, NetworkGlobalID, PublicKey
from ton_connect import AppWalletsLoader, FileStorage, TonConnect, VerifySignData
from ton_connect.models import (
SignDataPayloadDto,
SignDataPayloadText,
)
# Setup constants — see Installation and Initialization
MANIFEST_URL = "https://raw.githubusercontent.com/nessshon/ton-connect/main/assets/tonconnect-manifest.json"
SESSION_KEY = "user-123"
STORAGE_PATH = "./tonconnect-storage.json"
# Domain your app is served from — must match your tonconnect-manifest.json
APP_DOMAIN = "raw.githubusercontent.com"
# Toncenter API for the network the example connects to (testnet)
TONCENTER_API = "https://testnet.toncenter.com/api/v2"
# Text the user will be asked to sign in their wallet UI
SIGN_TEXT = "I agree to the Terms of Service"
async def get_wallet_public_key(address: Address) -> PublicKey | None:
# Current key from the get_public_key get-method; None falls back to the initial key
body = {"address": address.to_str(), "method": "get_public_key", "stack": []}
timeout = aiohttp.ClientTimeout(total=15)
async with (
aiohttp.ClientSession(timeout=timeout) as session,
session.post(f"{TONCENTER_API}/runGetMethod", json=body) as resp,
):
if not resp.ok:
return None
result = (await resp.json())["result"]
if result["exit_code"] != 0 or not result["stack"]:
return None
return PublicKey(result["stack"][0][1])
async def main() -> None:
storage = FileStorage(STORAGE_PATH)
app_wallets_loader = AppWalletsLoader(require_sign_data=True)
tc = TonConnect(
storage=storage,
manifest_url=MANIFEST_URL,
app_wallets=app_wallets_loader.get_wallets(),
)
connector = tc.create_connector(SESSION_KEY)
# Try to restore a previously saved connection before starting a new one
restored = await connector.restore()
if not restored:
request = connector.make_connect_request()
await connector.connect(
request=request,
network=NetworkGlobalID.TESTNET,
)
for app_wallet in app_wallets_loader.get_wallets():
print(f"{app_wallet.name}: {connector.make_connect_url(request, app_wallet)}")
wallet, error = await connector.wait_connect()
if error:
print(f"Connection failed: {error}")
return
address = connector.account.address.to_str(is_bounceable=False)
print(f"Address: {address}")
# Build the sign data payload
payload = SignDataPayloadText(text=SIGN_TEXT)
# Send the sign data request to the connected wallet
# Returns a request_id immediately — the user approves in their wallet app
request_id = await connector.sign_data(payload)
print("Sign request sent, waiting for confirmation...")
# Block until the wallet signs the data (or rejects/times out)
result, error = await connector.wait_sign_data(request_id)
if error:
print(f"Signing failed: {error}")
return
print(f"Signature: {result.signature.as_hex}")
print(f"Signed at: {result.timestamp}")
try:
# Verify the signature returned by the wallet
sign_data_payload = SignDataPayloadDto(
address=connector.account.address,
network=connector.account.network,
public_key=connector.account.public_key,
wallet_state_init=connector.account.state_init,
signature=result.signature,
timestamp=result.timestamp,
domain=result.domain,
payload=result.payload,
)
# get_wallet_public_key: optional, checks the Telegram wallet against its current key
await VerifySignData(sign_data_payload).verify(
allowed_domains=[APP_DOMAIN],
valid_auth_time=5 * 60, # 5 minutes
get_wallet_public_key=get_wallet_public_key,
)
print("SignData verified")
except BadSignatureError as e:
print(f"SignData verification failed: {e}")
# Close all active bridge connections
await tc.close_all()
if __name__ == "__main__":
import asyncio
asyncio.run(main())