Websockets

Introduction

Apart from interacting with Starknet by request-response model, you can also rely on real-time notifications. Here comes WebsocketClient which allows to establish a connection with Starknet node and listen for events.

Connecting

To begin interacting with Starknet via websockets, create a new instance of WebsocketClient and connect to the node.

from starknet_py.net.websockets.websocket_client import WebsocketClient

# Create `WebsocketClient` instance
websocket_client = WebsocketClient("wss://your.node.url")

# Connect to the websocket server
await websocket_client.connect()

Different subscription methods

New block headers

To subscribe to new block headers, use subscribe_new_heads(). Every time a new block is created, the event will be fired.

from starknet_py.net.websockets.models import NewHeadsNotification

# Create a handler function that will be called when a new block is created
def handler(new_heads_notification: NewHeadsNotification):
    # Perform the necessary actions with the new block...

# Subscribe to new heads notifications
subscription_id = await websocket_client.subscribe_new_heads(handler=handler)

# Here you can put code which will keep the application running (e.g. using loop and `asyncio.sleep`)
# ...

# Unsubscribe from the notifications
unsubscribe_result = await websocket_client.unsubscribe(subscription_id)

New events

To subscribe to new events, use subscribe_events(). Every time a new event is emitted, the event will be fired.

It’s possible to filter events by contract addresses, keys and block id. See all options in the method documentation.

from starknet_py.net.websockets.models import NewEventsNotification

# Create a handler function that will be called when a new event is emitted
def handler(new_events_notification: NewEventsNotification):
    # Perform the necessary actions with the new event...

# Subscribe to new events notifications
subscription_id = await websocket_client.subscribe_events(
    handler=handler, from_address=account.address
)

# Here you can put code which will keep the application running (e.g. using loop and `asyncio.sleep`)
# ...

# Unsubscribe from the notifications
unsubscribe_result = await websocket_client.unsubscribe(subscription_id)

Transaction status

To subscribe to transaction status changes, use subscribe_transaction_status(). Every time a transaction status changes, the event will be fired.

from starknet_py.net.websockets.models import TransactionStatusNotification

# Create a handler function that will be called when a new transaction status is emitted
def handler(transaction_status_notification: TransactionStatusNotification):
    # Perform the necessary actions with the new transaction status...

# Subscribe to transaction status notifications
subscription_id = await websocket_client.subscribe_transaction_status(
    handler=handler, transaction_hash=execute.transaction_hash
)

# Here you can put code which will keep the application running (e.g. using loop and `asyncio.sleep`)
# ...

# Unsubscribe from the notifications
unsubscribe_result = await websocket_client.unsubscribe(subscription_id)

Pending transactions

To subscribe to pending transactions, use subscribe_pending_transactions(). Every time a new pending transaction is added, the event will be fired.

It’s possible to filter pending transactions by sender address.

from starknet_py.net.websockets.models import PendingTransactionsNotification

# Create a handler function that will be called when a new pending transaction is emitted
def handler(pending_transaction_notification: PendingTransactionsNotification):
    # Perform the necessary actions with the new pending transaction...

# Subscribe to pending transactions notifications
subscription_id = await websocket_client.subscribe_pending_transactions(
    handler=handler,
    sender_address=[account.address],
)

# Here you can put code which will keep the application running (e.g. using loop and `asyncio.sleep`)
# ...

# Unsubscribe from the notifications
unsubscribe_result = await websocket_client.unsubscribe(subscription_id)

Handling chain reorganization notifications

When subscribing to new block headers, events, or transaction statuses, you also receive notifications of chain reorganizations. For example, if two blocks are produced nearly simultaneously and one replaces the other as the canonical block, you’ll get an update indicating that the chain has restructured. To handle them, you need to set the on_chain_reorg to your custom function.

from starknet_py.net.websockets.models import ReorgNotification

# Create a handler function that will be called when a reorg notification is emitted
def handler_reorg(reorg_notification: ReorgNotification):
    # Perform the necessary actions with the reorg data...

# Set the handler function for reorg notifications
websocket_client.on_chain_reorg = handler_reorg

# Subscribe to new heads, events, or transaction status notifications.
# In this example we will subscribe to new heads notifications.
# Note: we're passing empty handler function here, as we're only interested in reorg notifications.
subscription_id = await websocket_client.subscribe_new_heads(handler=lambda _: _)

# Here you can put code which will keep the application running (e.g. using loop and `asyncio.sleep`)
# ...

# Unsubscribe from the notifications
unsubscribe_result = await websocket_client.unsubscribe(subscription_id)

Disconnecting

To disconnect from the node, use disconnect().

# Disconnect from the websocket server
await websocket_client.disconnect()