WorkerBee (Python)

Python port of @hiveio/workerbee, a Hive blockchain automation library built on hiveio-wax.

WorkerBee provides an event-based observer pattern for bots that react to blockchain events such as posts, votes, transfers, account changes, market activity, and governance signals. It supports live streaming and historical replay.

Requirements

• Python 3.12+
• hiveio-wax 2.0.2.dev20260720133734

Installation

pip install hiveio-workerbee

Post-release development builds used by this repository are published in the Hive GitLab package registry. If pip cannot find hiveio-workerbee on the configured package index, install it with the Hive registry enabled:

pip install --extra-index-url https://gitlab.syncad.com/api/v4/groups/136/-/packages/pypi/simple hiveio-workerbee

Quick Start

import asyncio

from wax import WaxChainOptions, create_hive_chain
from workerbee import WorkerBee


async def main():
    async with (
        create_hive_chain(WaxChainOptions(endpoint_url="https://api.hive.blog/")) as chain,
        WorkerBee(chain) as bot,
    ):
        bot.observe.on_block().subscribe(on_next=lambda data: print(data["block"]["number"]))
        await bot.run_forever()


asyncio.run(main())

Use async iteration when the event handler needs to await work:

from wax import WaxChainOptions, create_hive_chain
from workerbee import WorkerBee


async def watch_account(account: str) -> None:
    async with (
        create_hive_chain(WaxChainOptions(endpoint_url="https://api.hive.blog/")) as chain,
        WorkerBee(chain) as bot,
    ):
        async for event in bot.observe.on_impacted_accounts(account):
            impacted = event["impacted_accounts"][account]
            print(f"{account} was impacted by {len(impacted)} operation(s)")

Architecture

Block Stream
    |
    v
+---------------+     +---------------+     +--------------+
|  Classifiers  | --> |  Collectors   | --> |  Factories   |
| (13 classes)  |     | (15 classes)  |     | (JsonRpc,    |
|               |     |               |     |  HistoryData)|
+---------------+     +---------------+     +--------------+
                                                   |
                                                   v
                                          +------------------+
                                          | ObserverMediator |
                                          +------------------+
                                            |             |
                                            v             v
                                      +---------+   +----------+
                                      | Filters |   | Providers|
                                      | (22)    |   | (22)     |
                                      +---------+   +----------+
                                            \           /
                                             v         v
                                          +-------------+
                                          |  QueenBee   |
                                          | (fluent API)|
                                          +-------------+
                                                |
                                                v
                                          +-------------+
                                          |  WorkerBee  |
                                          | (bot entry) |
                                          +-------------+

Pipeline: Classifiers declare required data, collectors fetch it, factories build the evaluation context, ObserverMediator coordinates filter matching and provider enrichment, and QueenBee delivers typed notifications to subscribers.

API Reference

WorkerBee

WorkerBee is the main bot class. It manages lifecycle, block streaming, transaction broadcast, and historical replay entry points.

from datetime import timedelta

from workerbee import WorkerBee, __version__

print(__version__)


async def stream_blocks(chain) -> None:
    async with WorkerBee(chain) as bot:
        async for block in bot:
            print(block["number"], block["id"])


async def broadcast_transaction(bot: WorkerBee, transaction) -> None:
    transaction_id = str(transaction.id)
    await bot.broadcast(transaction, expire_in=timedelta(seconds=30), verify_signatures=True)
    print(f"transaction {transaction_id} was seen on chain")


async def park_bot(bot: WorkerBee) -> None:
    await bot.run_forever()

Block iteration raises pipeline errors by default. To handle errors and keep iterating, pass a callback:

async def stream_with_error_handler(bot: WorkerBee) -> None:
    async for block in bot.iterate(on_error=log_error):
        print(block["number"])

QueenBee (bot.observe)

QueenBee is the fluent observer builder exposed through bot.observe. Each access returns a fresh builder, and a builder can be subscribed once.

Callbacks may be synchronous or async def:

bot.observe.on_posts("alice", "bob").subscribe(
    on_next=lambda data: print(data["posts"]),
    on_error=lambda err: print(f"observer error: {err}"),
)

bot.observe.on_votes("alice").subscribe(on_next=handle_vote)

bot.observe.on_accounts_balance_change(True, "alice").subscribe(on_next=handle_balance)

bot.observe.on_posts("alice").and_.on_votes("alice").subscribe(on_next=handle)

bot.observe.on_whale_alert({"amount": "100000", "nai": "@@000000021", "precision": 3}).subscribe(on_next=handle)

bot.observe.on_alarm("alice").subscribe(on_next=handle_alarm)

bot.observe.on_exchange_transfer().subscribe(on_next=handle_exchange)

bot.observe.on_custom_operation("splinterlands").subscribe(on_next=handle_custom)

Observer chains are also async-iterable:

async def consume_posts(bot: WorkerBee) -> None:
    async for event in bot.observe.on_posts("alice"):
        await handle_post(event)

The subscription closes when the iterator is finalized. For deterministic early shutdown, use callback subscriptions and sub.close(), or call aclose() on the async iterator object.

Available observer methods include:

on_block, on_block_number, on_transaction_ids, on_impacted_accounts, on_accounts_balance_change, on_accounts_metadata_change, on_accounts_full_manabar, on_accounts_manabar_percent, on_new_account, on_alarm, on_custom_operation, on_posts, on_comments, on_votes, on_mention, on_reblog, on_follow, on_posts_incoming_payout, on_comments_incoming_payout, on_feed_price_change, on_feed_price_no_change, on_whale_alert, on_exchange_transfer, on_internal_market_operation, and on_witnesses_missed_blocks.

Provider-only methods include:

provide_accounts, provide_witnesses, provide_rc_accounts, provide_block_header_data, provide_block_data, provide_feed_price_data, and provide_manabar_data.

Typed Notifications

Subscriber callbacks receive an ObserverNotification, a TypedDict(total=False) exported from workerbee.

from workerbee import ObserverNotification


def handle_event(event: ObserverNotification) -> None:
    accounts = event.get("accounts")
    if accounts is not None:
        print(accounts.keys())

Payload leaves that come directly from Hive APIs use canonical hiveio_api/wax models. WorkerBee-specific projections and grouping containers are typed in workerbee.chain_observers.payloads.

Application Boundaries

WorkerBee observes live blocks and bounded historical block ranges. It does not replace every Hive API that an application may need. For example, a bot that starts with an existing backlog of recent posts plus their active_votes should query Bridge directly with bridge.get_account_posts, then use WorkerBee for new live notifications.

WorkerBee also does not manage private keys. Build and sign transactions with Wax/Beekeeper, then pass the signed transaction to await bot.broadcast(tx). broadcast() returns None after the transaction has been seen on chain. If an application wants to log the transaction id, read str(tx.id) from the transaction object before or after calling broadcast().

Hive vote operation weights are basis points, not percentages: 10000 is a 100% vote and 1000 is a 10% vote. Application configuration that stores a percentage should convert it before constructing the Wax vote operation, for example weight=vote_weight_percent * 100.

on_accounts_full_manabar(...) mirrors the TypeScript WorkerBee API and means on_accounts_manabar_percent(..., 98, ...). Applications that need a stricter threshold, such as 99.98, should call on_accounts_manabar_percent(...) directly.

PastQueen

PastQueen processes historical blocks over a finite or open-ended replay range. Historical subscriptions complete when replay is exhausted. To continue with live observation, close the replay subscription and create a new live subscription from bot.observe on the same WorkerBee.

from workerbee.chain_observers import PastQueen

past = PastQueen(bot, from_block=80000000, to_block=80000100)
past.on_posts("alice").subscribe(on_next=handle_historical_post)

WorkerBee exposes two replay entry points, split by whether a chain query is needed:

def replay_range(bot) -> None:
    past = bot.provide_past_operations(80000000, 80000100)
    past.on_posts("alice").subscribe(on_next=handle_historical_post)


async def replay_recent(bot) -> None:
    past = await bot.provide_past_operations_relative("-1h")
    past.on_posts("alice").subscribe(on_next=handle_historical_post)

Examples

See examples/ for runnable scripts:

• block_parser.py - iterate over live blocks, print block id and number
• account_observer.py - monitor operations impacting an account
• post_observer.py - watch for new posts and sign votes when a private key is configured
• custom_json_spotter.py - print transactions from blocks containing custom_json_operation

Run an example:

python -m examples.block_parser --api-endpoint https://api.hive.blog/

Development

Requires Poetry 2.1.3+ with the poetry-dynamic-versioning plugin.

poetry install

poetry run pre-commit run --all-files --config ../.pre-commit-config.yaml

poetry run pytest tests/unit/test_import_time.py -q
poetry run pytest tests/unit -q -n auto

Integration tests run against the configured mirrornet endpoint:

WORKERBEE_MIRRORNET_ENDPOINT=https://api.hive.blog poetry run pytest tests/integration -q -n auto
poetry run pytest tests/mock -q

Agent/developer practices for the port live in AGENTS.md.

Relationship to TypeScript Codebase

This is a full port of the TypeScript @hiveio/workerbee library. It preserves:

• 4-layer architecture: Classifier -> Collector -> Factory -> ObserverMediator
• class names such as WorkerBee, QueenBee, PastQueen, and ObserverMediator
• filter/provider pairing logic
• fluent API composition with and_ and or_

The TypeScript source under ../src/ is the behavioral source of truth. Python public names use snake_case for PEP 8 compliance, for example onBlock() becomes on_block() and provideBlockData() becomes provide_block_data().

The Python port depends on hiveio-wax instead of the npm @hiveio/wax package.

License

See repository root for license information.