AlgoKit Subscriber Python

Installation

pip

pip install algokit-subscriber

uv

uv add algokit-subscriber

Quick Example

import algokit_subscriber as sub
from algokit_utils import AlgorandClient

algorand = AlgorandClient.testnet()

subscriber = sub.AlgorandSubscriber(
    config=sub.AlgorandSubscriberConfig(
        filters=[
            sub.SubscriberConfigFilter(
                name="filter1",
                filter=sub.TransactionFilter(
                    type="pay",
                    sender="ABC...",
                ),
            ),
        ],
        watermark_persistence=sub.in_memory_watermark(),
        sync_behaviour="skip-sync-newest",
    ),
    algod_client=algorand.client.algod,
)

# Set up subscription(s)
subscriber.on("filter1", lambda transaction, _: print(transaction.id_))

# Set up error handling
subscriber.on_error(lambda error, _: print(error))

# Either: Start the subscriber (if in long-running process)
subscriber.start()

# OR: Poll the subscriber (if in cron job / periodic lambda)
subscriber.poll_once()

Features

Real-time Subscriptions

Subscribe to transactions at the tip of the chain with low-latency processing.

Indexing & Catchup

Build indexes from the beginning of the chain with fast indexer catchup.

ARC-28 Events

Subscribe to and process ARC-28 events from smart contract calls.

Resilient Watermarking

Persist watermarks for reliable at-least-once delivery across outages.