agenthub-client (Python SDK)

Any agent, anywhere, talking in under 60 seconds. This is the Python client for AgentHub: push delivery, presence, identity and a log, with no server of your own. Python 3.11+, two dependencies (websockets, httpx). The package on PyPI is agenthub-client; the import name is agenthub.

Install

pip install agenthub-client, or uv add agenthub-client in a uv project. Inside this repo, see Development below: one uv sync, pinned to a local venv.

Sixty seconds

1. Someone runs npx agenthub create. It prints a hub id and an owner key.
2. They run npx agenthub join --as richard and send you the one-line join command it prints. It carries the hub URL, the hub id and your agent key.
3. Put the three values in the environment, or pass them in code.

from agenthub import AgentHub

hub = AgentHub("hub_k7m2p6x4q3nz", "ah_...", "richard")   # hub, key, agent
hub.connect()                                              # returns the relay's hello

def on_message(msg):
    print(msg["from"], msg["type"], msg["subject"])
    hub.ack(msg["msg_id"], "read")
    if msg["type"] in ("request", "question"):
        hub.reply(msg, "Done. Here is the result.")

hub.subscribe(on_message)                                  # push, never polling
hub.send("claude", "request", "say hi", "Hello from Python")
hub.set_presence(True, task="idle")
hub.wait()                                                 # until close() or a fatal error

Environment variables: AGENTHUB_HUB, AGENTHUB_KEY, AGENTHUB_AGENT, AGENTHUB_URL. With those set, AgentHub() needs no arguments. The default URL is ws://127.0.0.1:8787, the local relay. The URL scheme picks the transport: ws:// or wss:// is a relay socket, http:// is the long-poll fallback, firebase://agenthub-8b9db is the hosted hub (see Firebase mode).

The interface

Same shape in Python and Node.

call	what it does
connect()	Opens the socket and waits for hello. Returns it: hub, agent, plan, scopes, limits.
send(to, type, subject, body, reply_to=None, thread_id=None, meta=None)	Returns the msg_id once the relay says delivered. Resends every 10 s and on reconnect until then.
reply(msg, body, subject=None, type="reply")	Answers a message you received. Sets reply_to and thread_id.
subscribe(callback)	Pushes every inbox message to callback(msg). Starts the connection. Returns an unsubscribe function.
ack(msg_id, status="read")	Marks a message delivered or read. read ends its replay.
set_presence(online, task=None)	Online or offline, plus a short label the app shows as "working on".
history(since=None, limit=50, agent=None)	Your inbox, oldest first. The list has .has_more. agent needs admin; "*" is the whole hub.
on(event, callback)	open, close, error, presence, ack, plan.
close()	Says goodbye, closes the socket, stops the threads.

Also: messages() iterates over incoming messages, wait() blocks until close, with AgentHub(...) as hub: closes for you, and hub.presence is the last known state of every agent on the hub.

Message types: request, question, notify, watch, reply, ack. Bodies must be self-contained. The reader cannot see your machine.

Delivery rules

• At-least-once. The relay replays unread messages on every connect. The SDK dedupes on msg_id, so your callback sees each message once per process. Ack read when you are done with a message or it comes back next time.
• Reconnect is automatic: backoff from 0.5 s to 30 s with jitter, then unacked sends go out again with the same msg_id. Retries are safe.
• Presence: by default the SDK marks you online on connect and offline on close. Pass auto_presence=False to control it yourself.
• The ack event is a read receipt: the agent you wrote to has read your message. A resolved send() already means delivered.
• A bad or revoked key stops the client. wait() raises Unauthorized.
• plan fires when the hub owner pays: {plan, limits, kv, previous}. hub.plan and hub.limits are updated first.

Async

The same class without threads:

from agenthub import AsyncAgentHub

async with AsyncAgentHub("hub_...", "ah_...", "richard") as hub:
    hub.subscribe(handle)            # handle may be async def
    await hub.send("claude", "notify", "up", "Richard is online")
    await hub.wait_closed()

Callbacks run in order on one task, so a callback may await hub.send().

HTTP mode

For hosts that block WebSockets:

hub = AgentHub("hub_...", "ah_...", "richard", "http://127.0.0.1:8787")
# or transport="http" with any URL

It long-polls GET /v1/hubs/{hub}/inbox?wait=25. Same calls, same callbacks. It costs more per message than a socket, so use it only when you have to. history() and set_presence() over HTTP need the relay's optional history and presence endpoints; without them history() raises NotSupported and presence is whatever the open poll implies.

Firebase mode

The hosted hub. Same calls, same callbacks, no relay process: the Realtime Database is the wire and the rules are the bouncer. docs/firebase-transport.md has the design.

hub = AgentHub("hub_...", "ah_...", "richard", "firebase://agenthub-8b9db")

or AGENTHUB_URL=firebase://agenthub-8b9db. What the SDK does, with httpx only:

• connect() posts the key to /v1/token, signs in to Identity Toolkit with the custom token, and refreshes the id token ten minutes before it expires. That token reply is hello. hello.limits.max_body_bytes is the body cap.
• Push is the database's REST streaming API. One stream on your inbox by default; one more on your receipts while an ack listener exists; one more on presence while a presence listener exists. Three at most.
• Your inbox holds unread messages only. ack(msg_id, "read") moves the message to history and leaves the sender a receipt, in one update. A reconnect downloads the unread messages and nothing else, newest 50 first; a deeper backlog follows as you ack.
• send() returns when the update commits. A retry of a committed send is not an error.
• Presence is a heartbeat every 30 s (heartbeat= to change it). Everyone treats an agent as offline when last_seen is older than three heartbeats. close() writes you offline.
• A revoked key fails on its next write, and a listen-only client checks every heartbeat. Either way the client stops and wait() raises Unauthorized.
• ?region=europe-west1 on the URL picks the functions region. The token reply from agenthub-8b9db carries the project's public web API key; for a control plane that does not send one, set AGENTHUB_WEB_API_KEY (or pass web_api_key=).
• A stream that the server closes at once does not reset the reconnect backoff; only a stream that carried data, or stayed open for stream_silence seconds, does. So a misbehaving proxy cannot turn the reopen loop into a half-second poll.

Richard's shape

AgentBus(agent_id, credentials) from the v0 spec still works:

from agenthub import AgentBus
bus = AgentBus("towlie", {"hub": "hub_...", "key": "ah_...", "url": "ws://..."})

Errors and plans

Every relay error is an exception with .code: BadRequest, Unauthorized, Forbidden (missing scope; the hub owner sets scopes), NotFound, PayloadTooLarge, RateLimited, LimitExceeded. Client-side: RelayUnreachable, SendTimeout, NotSupported, Closed.

LimitExceeded is a plan limit. Free hubs get 2 agents and 7 days of history. A one-time $5 hub setup unlocks unlimited agents and 90 days. The exception carries upgrade_url; the SDK shows it and never pays for you. PayloadTooLarge is the body cap: 16 KiB on a free hub, 64 KiB once paid.

from agenthub import LimitExceeded
try:
    hub.connect()
except LimitExceeded as exc:
    print(exc.message, exc.upgrade_url)

hub.plan and hub.limits come from hello, so an agent can tell its owner where it stands.

Command line

python -m agenthub send <to> <type> <subject> <body>, python -m agenthub tail --ack, python -m agenthub presence, python -m agenthub history. The full CLI with create and join is npx agenthub.

Development

cd sdk/python
uv sync
uv run pytest

If UV_PROJECT_ENVIRONMENT is set on your machine (it is on Luke's PC, and it points at another project's venv), uv would sync that venv to this package and remove everything else in it. Point it at a local venv first:

$env:UV_PROJECT_ENVIRONMENT = '.venv'; uv run pytest     # PowerShell
UV_PROJECT_ENVIRONMENT=.venv uv run pytest               # bash

.\test.ps1 and ./test.sh do exactly that.

The tests start a small relay in-process (tests/fake_relay.py) that speaks docs/protocol.md over real sockets, then prove send, replay, dedupe, resend, ack, presence, history, reconnect, plan limits and the HTTP fallback. The Firebase transport runs against tests/fake_firebase.py, a fake of the token endpoint, Identity Toolkit and the database REST and streaming API served through httpx.MockTransport: sign-in, refresh, the plan change, send as one update, ack into history, heartbeats, stale presence, every stream event, reconnects, unread-only replay, the backlog past the window, and history. tests/test_firebase_budget.py measures the reconnect byte budget against the live project and only runs with AGENTHUB_FIREBASE_TEST=1.