AlgoKit Subscriber Python

Installation

pip

pip install algokit-subscriber

uv

uv add algokit-subscriber

Quick Example

from algokit_subscriber import AlgorandSubscriber
from algokit_utils import get_algod_client, get_algonode_config

# Create an Algod client (testnet used for demo purposes)
algod_client = get_algod_client(get_algonode_config("testnet", "algod", ""))

subscriber = AlgorandSubscriber(
    config={
        "filters": [
            {
                "name": "filter1",
                "filter": {
                    "type": "pay",
                    "sender": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAY5HFKQ",
                },
            },
        ],
        "watermark_persistence": {"get": lambda: 0, "set": lambda x: None},
        "sync_behaviour": "skip-sync-newest",
        "max_rounds_to_sync": 100,
    },
    algod_client=algod_client,
)

# Set up subscription(s)
subscriber.on("filter1", lambda transaction, _: print(f"Received transaction: {transaction['id']}"))

# Set up error handling
subscriber.on_error(lambda error, _: print(f"Error occurred: {error}"))

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

# OR: Poll the subscriber (if in cron job / periodic lambda)
result = subscriber.poll_once()
print(f"Polled {len(result['subscribed_transactions'])} transactions")

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.