Session events and listeners

Workflow.session() returns a Session, a handle on one session of a workflow revision. It reads the session’s event stream, lists its listeners and switches them, and lists the executions a listener started. Everything goes through the revision’s data channel, so the SDK opens no WebSocket: following the stream is polling.

import blobhub
from blobhub.types.workflow import Workflow

hub = blobhub.connect()
workflow = Workflow(hub.blob("acme", "support-bot").revision())

session = workflow.session("sdk-demo")     # reads the session now; never creates one
for event in session.events({"types": ["thread.item_posted"]}, limit=20):
    print(event["created_at"], event["type"], event["object"]["alias"])

Events are plain dict records, as everywhere in the SDK. The SDK checks no event type: the taxonomy is the platform’s, so a new type needs no SDK release.

The session handle

workflow.session(session, *, verify=True) takes a session id or alias.

`verify`	What happens
`True`, the default	Reads the session, so one that is absent or unreachable fails here with `PermissionDenied`. The handle keeps the record.
`False`	Sends nothing and holds only the name given. It is how to read the stream of a session that was deleted: that stream is still served by id while it lasts.

It never creates a session, unlike execute(session=...).

Attribute	Holds
`id`	The record’s id, or the name given when unverified.
`alias`, `status`, `record`	From the record read; `None` when none was read.

session.refresh() reads the record again. A handle that holds a record sends its id on every read, so one opened by alias keeps reading the same session even if that session is deleted and the alias is reused. An unverified handle sends the name it was given. A finished run leads back to its session: run.session is the Session for the execution’s session_id, and run.source says what started it.

Reading events

session.events(filter=None, *, after=None, before=None, ascending=True, limit=None, page_size=100, follow=False, interval=2.0, timeout=None)

returns a lazy iterator over the stream.

Argument	Meaning
`filter`	`{types?, aliases?, executions?, where?}`, the event filter. The server checks it; a refusal is `CommandError` (`invalid_filter`) naming the clause.
`after`, `before`	Exclusive event ids that bound the range. Build one from a time with `boundary_id`.
`ascending`	Oldest first, the default. The platform’s own default is newest first, so the SDK reverses it; pass `False` to read newest first.
`limit`	The total number of events to yield, then stop.
`page_size`	Events per request, 1 to 1000.
`follow`, `interval`, `timeout`	See Following.

Each request is one list_session_events; the iterator follows the platform’s cursor until it is null, and never stops on a short page, because the filter runs on the server and a page can hold fewer matches than page_size while more remain. Arguments the SDK can check without a request are checked before the first one: a bad page_size, limit, after, before or filter is ValueError. The filter’s own grammar is the server’s.

# the five newest events that touched a worker object
newest = list(session.events({"aliases": ["worker/**"]}, ascending=False, limit=5))

session.latest() returns the newest event, or None for an empty stream, in one request. session.event(event_id) returns one event and raises NotFound when there is none.

Following

follow=True keeps the iterator open. When the range is exhausted it sleeps interval seconds and reads again, until the caller closes the iterator, limit events were yielded, or timeout seconds have passed. The deadline ends the follow whether or not anything remains to read. Following needs ascending=True and no before; either combination is ValueError.

start = session.latest()
after = start["id"] if start else None
for event in session.events({"types": ["object.written"]}, after=after, follow=True, timeout=300):
    print(event["object"]["alias"])

A late commit is not missed. An event is visible to a read at most 60 seconds after the time in its id, so after: <last id> alone can skip a row that commits late. Each pass of a follow therefore re-reads from a bound a minute behind the newest id it knew before the previous pass began, and drops the ids it already yielded. See Settle time. Two consequences:

Events arrive in id order except a late one, which arrives when it is found, after events with larger ids. No event is yielded twice.
A quiet stream re-reads the same last minute every interval: one small listing a poll. Raise interval for a slow reader. A clock skew between the platform’s producers larger than the window can still hide a row.

A follow ends with the error when the session goes away: the next read raises PermissionDenied.

Boundary ids

boundary_id(moment) returns the id that sorts before every event of moment’s microsecond, so after=boundary_id(t) starts at t. It takes epoch seconds or an aware datetime; a naive datetime is ValueError, because the SDK never guesses a zone.

from datetime import datetime, timezone
from blobhub.types.workflow import boundary_id

since = boundary_id(datetime(2026, 11, 9, 9, 0, tzinfo=timezone.utc))
events = list(session.events(after=since))

execution.ended

run.wait() ends on the run’s status, not on this event. To read the event itself:

ended = next(session.events({"types": ["execution.ended"], "executions": [run.id]}), None)

It carries the run’s status, and its route and error when there are any. The platform writes it best-effort, so it is usually there but not guaranteed; the run’s status is the authority.

Listeners

session.listeners(*, binding=None) returns the session’s listeners as the platform reports them: every declared listener of the revision, then the connection listeners open on the session. binding="declared" or "connection" keeps only that kind, and anything else is ValueError.

for entry in session.listeners(binding="declared"):
    print(entry["alias"], entry["enabled"], entry["counts"], entry["in_flight"])

A declared entry carries definition_id, alias, description, enabled (the effective switch), session_enabled (present only when the session overrides the definition), definition, counts, last_outcome, outcomes (the newest 20), in_flight and, when a batch is collecting, pending. A connection entry carries listener_id, connection_id, filter, batch and created_at. See Listeners for what the outcomes mean. session.set_listener(listener, enabled) switches a declared listener on or off for this session and returns its entry. listener is an alias or definition id and enabled must be a bool; anything else is TypeError or ValueError before any request. An unknown listener is NotFound.

entry = session.set_listener("orchestrate", False)
print(entry["enabled"], entry.get("session_enabled"))

The switch works on a session of any status and raises no event. Setting a value equal to the definition’s own clears the override. It is a write, so the SDK repeats it only when the failure proves it never ran (see Errors and retries); after a failure that may have reached the platform, read listeners() to learn the state, or set it again, which is idempotent. session.executions(listener=None) yields the session’s executions as Execution objects, newest first, following the platform’s cursor to its end. With listener (an alias or definition id) it yields only the runs that listener started; their run.source["kind"] is listener.

for run in session.executions(listener="orchestrate"):
    print(run.id, run.status, run.source["listener"]["alias"])

A listener alias that resolves to no listener of the revision is PermissionDenied, unless it is shaped like an id: a deleted listener’s id still lists the executions it started. Creating, uploading and editing a listener definition is blobhub-cli’s job, as for workflow definitions. A connection listener needs a WebSocket, which the SDK does not open; use the WebSocket listener protocol.

Errors

Error	When
`ValueError`	A bad `page_size`, `limit`, `after`, `before`, `filter`, `interval`, `binding`, `listener` or `event_id`; a naive `datetime` for `boundary_id`; `ascending=False` or `before` with `follow`.
`TypeError`	`enabled` is not a `bool`, or `listener` is not a string.
`PermissionDenied`	The session id or alias is absent or not reachable with this key, or a listener alias for `executions` resolves to nothing.
`NotFound`	`event()` finds no such event, or `set_listener` names no declared listener.
`CommandError` `invalid_filter`	The filter breaks the grammar or its bounds.
`CommandError` `invalid_cursor`	The platform refused a cursor.
`ServerError` `cursor_loop`	The platform returned a cursor the SDK had already sent.

Python SDK