Developer advocate from Chainstack here. This is a sample code to subscribe to logs for the token program:
Python - 3.9.13
Solana - 0.29.1
base58 - 2.1.1
asyncio - 3.4.3
asyncstdlib - 3.10.5
import solana
import asyncio
from asyncstdlib import enumerate
from solana.rpc.websocket_api import connect
from solders.pubkey import Pubkey
from solders.rpc.config import RpcTransactionLogsFilterMentions
wss_url = ""
async def main():
async with connect(wss_url) as websocket:
await websocket.logs_subscribe(RpcTransactionLogsFilterMentions(Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")),commitment="finalized")
print("started")
first_resp = await websocket.recv()
subscription_id = first_resp[0].result
async for idx, msg in enumerate(websocket):
if idx == 10:
break
print(msg)
await websocket.logs_unsubscribe(subscription_id)
await main()
You need to fill in your wss endpoint in the code. If you need one, feel free to try out our services. Our developer plan is free and it gives 3 Million free requests.
