---
question: Does subscribe replay events published earlier?
kind: task
via_web: available
via_cli: na
via_python: available
via_api: available
via_config: na
ref_web: Events
ref_python: events.subscribe
ref_api: GET /api/admin/events/recent
prereq: For the Python steps, an in-engine module that lists `EventService` in its dependencies. For Nexus and the REST API, a super admin account.
next:
  - /capabilities/build/platform/record-an-event
  - /capabilities/build/platform/events-overview
---

# Stream events

A live stream shows events as they are published. In Python, `subscribe` calls you back for each new event and sends nothing published before the call. Nexus and the REST API poll the log instead.

## Web UI

Super admins only.

### Open the live feed

In Nexus, open **Events** in the sidebar. The feed lists events newest first and checks for new ones every 5 seconds. Select an event to see it drawn as a graph in the main pane.

![Events live feed](/abi/Screenshot_Events.png)

*Events: the live feed on the left, the process graph in the middle, and the selected event on the right.*

### Narrow it

Pick an event type to see only that type, or type in **Search payload (whole log)**: the search runs over every stored event, not only the loaded page. **Load older**, at the end of the list, pages back.

### Pause the tail

Choose the pause button (**Pause the live tail**) to keep the list still while you read. Polls that come in while it is paused are dropped. When you resume, the next poll brings back only the newest 100 events; reach the ones in between with **Load older**. **Clear loaded events** empties the list, not the log.

### Browse by type

**System** › **Data** › **Events** groups the log by event type, with a count and when each was last seen. Open a type to list its events, newest first, and search inside their payloads. The log is read-only there too.

## CLI

Not applicable. There is no `abi` command that follows events. Use another tab on this page. The command list is the [CLI reference](/user-guide/reference/cli).

## Python

### Subscribe

```python
from my_module.ontologies.modules.OrderEvents import OrderShipped

def on_shipped(event: OrderShipped) -> None:
    print(event.order_id)

self.engine.services.events.subscribe(OrderShipped, on_shipped)
```

The callback gets an instance of the class. Every subscriber gets its own copy; two subscribers on one type do not compete. `subscribe` returns the thread that listens. Keep the callback short: it runs on the bus consumer thread, and an exception in it is logged and skipped.

### Filter

```python
self.engine.services.events.subscribe(
    OrderShipped,
    on_shipped,
    filter={"order_id": {"prefix": "A-"}},
)
```

A filter key is a dotted path into the event. A plain value means equals, a list means one of, and an operator dict takes `eq`, `ne`, `prefix`, `suffix`, `contains`, `in`, `gt`, `gte`, `lt`, `lte` or `exists`. Several keys must all match.

### Catch up after a restart

`subscribe` replays nothing. A worker that reads the log with a cursor of its own picks up after a restart:

```python
pending = self.engine.services.events.query_for_consumer(
    "my_module:shipping-worker",
    OrderShipped,
    limit=100,
)
```

Each call returns the next events this consumer has not read and moves its cursor past them. `iter_query_for_consumer` drains them in batches. `seek_consumer_to_end` skips what is pending.

:::warning At most once
The cursor moves when the events are read, not when you finish them. If the worker crashes halfway through a batch, the rest of that batch is not returned again. Keep batches small (`limit`), and make the handling safe to repeat so you can re-read by `seq` with `query(since_seq=...)` after a crash.
:::

The cursor belongs to the consumer id and the event type, not to the filter. Events a filter skips are skipped for good for that consumer, so keep one filter per consumer id. With the `postgresql` adapter in NATS mode, a consumer idle longer than `archive_after_days` is left behind: the events it did not read move to the archive.

### Read the history

```python
last_50 = self.engine.services.events.query(OrderShipped, newest_first=True, limit=50)
```

`query` takes the same `filter`, plus `since_seq`, `until_seq`, `since_timestamp`, `until_timestamp` and `search` (text anywhere in the payload). `iter_query` walks a large range in batches, up to the last event when it started.

## REST API

The admin routes of the Nexus API. Auth: a Nexus session of a super admin (`POST /api/auth/login` returns the access token). There is no push stream: poll.

### List the event types

```bash
curl "http://localhost:9879/api/admin/events/types" \
  -H "Authorization: Bearer $NEXUS_TOKEN"
```

Each type has a `uri` and a `label`. With `abi dev up`, use the `api` port that `abi dev ports` prints.

### Poll the latest events

```bash
curl "http://localhost:9879/api/admin/events/recent?limit=100&event_class=http://ontology.naas.ai/abi/orders/OrderShipped" \
  -H "Authorization: Bearer $NEXUS_TOKEN"
```

It returns the last `limit` matching events (1 to 1000), oldest first. Each one carries `_seq`. There is no `since_seq`: keep the highest `_seq` you have seen and skip the events at or below it on the next poll. `q` searches the payloads. An `event_class` the API process has not loaded returns `[]`, not an error: take the `uri` from `/events/types`.

### Fill the gaps

When more than `limit` events arrive between two polls, the oldest of them are not in the reply. If the lowest `_seq` returned is still above the last one you saw, page back with `before_seq` set to that lowest `_seq`, and repeat until you reach it:

```http
GET /api/admin/events/recent?limit=100&before_seq=5120
```

Poll more often than events arrive in batches of `limit`, or raise `limit`.

## Config

Not applicable. A subscription is a call. `services.event` only picks where the log lives: see [Events](/user-guide/build/platform/events-overview#backends).