Alert Broker & Python SDK

The Astro-COLIBRI alert broker streams multi-messenger transient alerts in real time: gravitational waves, gamma-ray bursts, neutrinos, fast radio bursts, optical transients and more. It is an Apache Kafka broker, so you can consume it from any language that has a Kafka client, and alerts reach you within seconds of Astro-COLIBRI processing them rather than on a polling interval.

astrocolibri is the official Python SDK for that broker. It is a thin wrapper around confluent-kafka that handles the connection, authentication and consumer-group details for you. Using it is optional: everything it does is documented below, so you can equally point your own Kafka client at the broker.

Both are consume only. Publishing is restricted to the Astro-COLIBRI infrastructure, so no client can inject alerts into the streams.

In a hurry? Quick Start gets you your first alert in under a minute. To connect a non-Python client, see Connection parameters.

The Alert Broker

Getting access

Broker access is self-service. Sign in on astro-colibri.com, open your account page and choose Manage broker access. You are issued a personal username and password for SASL/SCRAM authentication.

The password is shown once, at creation: the broker stores only a SCRAM verifier and cannot reveal it again. If you lose it, rotate the credential from the same page. You can also revoke it there, which takes effect immediately.

Access requires a verified account and acceptance of the terms of use. Credentials identify you personally, so treat them as any other secret and keep them out of your repositories.

Connection parameters

Everything needed to point any Kafka client at the public broker:

Setting	Value
bootstrap.servers	broker.astro-colibri.science:9092
security.protocol	SASL_SSL
sasl.mechanism	SCRAM-SHA-512
sasl.username / sasl.password	the credentials from your account page
group.id	must begin with `<your-username>.` — see Access scope

The connection is TLS encrypted throughout. No plaintext listener is exposed to the internet.

Available topics

Topic	Format	Content
astrocolibri.all.JSON	JSON	Every alert published by Astro-COLIBRI
astrocolibri.all.VOEvent	VOEvent / XML	Every alert published by Astro-COLIBRI
astrocolibri.important.JSON	JSON	The subset that is scientifically actionable — see What reaches the important stream
astrocolibri.important.VOEvent	VOEvent / XML	The subset that is scientifically actionable — see What reaches the important stream
astrocolibri.heartbeat	JSON	Periodic liveness ping. It travels the same path as a real alert, so receiving it confirms the whole publishing pipeline is healthy, not merely that your connection is open

The same alert is published to both the JSON and the VOEvent topic, so choose the format you want rather than subscribing to both. Every topic has 3 partitions. From Python the current list is also available at runtime as astrocolibri.TOPICS.

What reaches the important stream

The important topics carry a subset of the all topics, never anything extra: every alert on important is also on all. Subscribe to important when you want the events worth reacting to without filtering the full stream yourself, and to all when you want a complete record.

Two things reach it:

• Every newly detected event, whatever its type. The first time Astro-COLIBRI publishes an event, it is always important.
• Updates that change the science of an event already published. An update qualifies only when it modifies at least one field that would change what you do about the event.

Updates that touch nothing in that set stay on the all topics only. That is what keeps the important stream usable: an event being revised a dozen times as bookkeeping, presentation or follow-up metadata is added does not repeatedly interrupt you.

What changed	Fields
Position or localisation	ra, dec, err, map_areas, gw_contours
Identity or classification	type, alert_type, classification, source_name, assoc
Distance or energetics	distance, distance_error, redshift, energy
Significance	far, signalness
Gravitational-wave classification	gw_BBH, gw_BNS, gw_NSBH, gw_MassGap, gw_Noise, gw_hasNS

A refined GW localisation, a neutrino gaining an association, a transient being reclassified or a burst having its redshift measured all qualify. Adding a light-curve figure or a follow-up note does not.

One further limit applies to updates: once an event is more than about a month old, its updates stay on the all topics whatever changed. Late refinements to a long-past transient are still published, they simply no longer interrupt the important stream. A newly detected event is never affected by this, even when the detection itself is historical.

Retention and delivery

Alerts are retained for 24 hours. The broker is a live stream rather than an archive, so to reach anything older use the REST API.

Your read position is tracked per consumer group on the broker, so a client that stops and restarts resumes where it left off and misses nothing published meanwhile, as long as the gap stays inside the retention window. Delivery is at least once: a client that dies between handling an alert and committing its offset will see that alert again, so make your processing idempotent if duplicates would matter to you.

Consumers sharing one group.id have the partitions divided between them, and each then receives only a share of the stream. That is what you want when scaling one pipeline across several replicas, and not what you want for two independent programs, which each need their own group.

Access scope

Your credentials are read-only and scoped by broker-side ACLs. They let you read and describe the astrocolibri.* topics, and use consumer groups whose name begins with <your-username>.

That group prefix is enforced by the broker, so a group.id without it is refused. The Python SDK adds the prefix for you; with any other client you have to include it yourself. No client is granted write access to any topic.

Using a client other than the SDK

The Python SDK is a convenience, not a requirement. Any Kafka client works with the parameters above. As a quick check from the command line, using kcat:

kcat -b broker.astro-colibri.science:9092 \
    -X security.protocol=SASL_SSL \
    -X sasl.mechanism=SCRAM-SHA-512 \
    -X sasl.username=alice \
    -X sasl.password=... \
    -G alice.kcat astrocolibri.all.JSON

Note the -G alice.kcat: the consumer group carries the username prefix the ACL requires. The same holds for the Java, Go, C# and other librdkafka-based clients.

Python SDK

Installation

pip install astro-colibri

Requires Python 3.9+. This installs confluent-kafka as a dependency, which requires the librdkafka C library. On most platforms this is bundled automatically with the wheel; if you hit a build error, install librdkafka via your system package manager first (e.g. apt install librdkafka-dev or brew install librdkafka).

Alternatively, install the latest version directly from the GitHub repository:

pip install git+https://github.com/astro-transients/astro-colibri.git

Quick Start

Create a Consumer with the credentials generated from your account, subscribe to one or more topics, and iterate over incoming alerts:

from astrocolibri import Consumer

with Consumer(
    username="your-username",
    password="your-password",
) as consumer:
    consumer.subscribe(["astrocolibri.all.JSON"])
    for message in consumer.consume(timeout=30):
        print(message.value())

The client connects to the official broker and starts from the earliest alert still inside the retention window. Your read position is remembered across restarts, so you resume exactly where you left off. If you run several scripts on the same credentials, give each one its own group_id (see Consumer() below).

Consumer Reference

CLASS

astrocolibri.Consumer(username, password, *, broker_url=None, group_id=None, start_at="earliest", security_protocol="SASL_SSL", config=None)

show/hide

Description

Subscribes to Astro-COLIBRI broker topics and receives alerts in real time, authenticated via SASL/SCRAM-SHA-512. Can be used directly or as a context manager (recommended, see example below).

Parameters

Parameter	Type	Default	Description
username	String	required	Username generated by the Astro-COLIBRI account GUI
password	String	required	Password generated alongside `username`
broker_url	String	broker.astro-colibri.science:9092	Kafka broker address. Use `"localhost:9092"` to point at a local broker for testing.
group_id	String	None	Kafka consumer group ID, prefixed with your username to satisfy the per-user ACL, so `"my-pipeline"` becomes `"alice.my-pipeline"`. Left unset, the group `"<username>.default"` is used. Kafka remembers your read position (offset) across sessions and you resume exactly where you left off. Set a `group_id` you have never used before to start reading from the beginning of the 24h retention window again. Running several scripts on the same credentials? Give each one its own `group_id`: consumers sharing a group have the partitions split between them, so each would otherwise receive only part of the stream.
start_at	String	"earliest"	Starting position for a new `group_id`. "earliest" receives every alert available within the 24h retention window; "latest" receives only new alerts published after connecting.
security_protocol	String	"SASL_SSL"	"SASL_SSL" for the official public broker (TLS-encrypted).
config	Dict	None	Extra `confluent-kafka` options merged into the base configuration. Authentication and consumer group identity remain managed by the client and cannot be overridden.

Raises

Exception	Condition
AstrocolibriConfigError	`start_at` is neither "earliest" nor "latest"
AstrocolibriAuthError	Connection to the broker fails (invalid credentials, network issue, etc.)

Examples

from astrocolibri import Consumer

# Official Astro-COLIBRI broker (default)
consumer = Consumer(
    username="alice",
    password="alice-strong-password",
)

# Local broker for testing
consumer = Consumer(
    username="alice",
    password="alice-strong-password",
    broker_url="localhost:9092",
    security_protocol="SASL_PLAINTEXT",
)

# Several scripts on the same credentials: one group_id each,
# so every script receives the full stream
ingest = Consumer(
    username="alice",
    password="alice-strong-password",
    group_id="ingest",
)

alerting = Consumer(
    username="alice",
    password="alice-strong-password",
    group_id="alerting",
)

METHOD

.subscribe(topics, *, on_assign=None, on_revoke=None, on_lost=None)

show/hide

Parameter	Type	Description
topics	List[String]	Topic names to subscribe to, e.g. `["astrocolibri.all.JSON"]`
on_assign	Callable	Callback `(consumer, partitions)` called when partitions are assigned
on_revoke	Callable	Callback `(consumer, partitions)` called when partitions are revoked
on_lost	Callable	Callback `(consumer, partitions)` called when partitions are lost involuntarily

METHOD

.consume(num_messages=1, timeout=-1)

show/hide

Description

Consumes messages from the subscribed topics and yields them one at a time. _PARTITION_EOF errors (end of partition) are silently ignored; any other Kafka error raises AstrocolibriKafkaError.

Parameters

Parameter	Type	Default	Description
num_messages	Integer	1	Maximum number of messages fetched per internal batch. Does not affect the stream itself: the generator always yields one message at a time.
timeout	Float	-1	-1 blocks indefinitely until the next message arrives; the generator never terminates on its own. A value > 0 returns after that many seconds with no message, terminating the generator — useful for periodic processing between batches.

Yields

confluent_kafka.Message, with useful methods:

Method	Returns	Description
message.value()	Bytes	Alert payload
message.key()	Bytes \| None	Message key
message.topic()	String	Source topic name
message.partition()	Integer	Kafka partition
message.offset()	Integer	Kafka offset
message.timestamp()	(type, timestamp_ms)	Message timestamp

Raises

Exception	Condition
AstrocolibriKafkaError	An unrecoverable Kafka error is encountered

Examples

Infinite loop (blocks until the next message):

import json

for message in consumer.consume():
data = json.loads(message.value())
print(data)

With a timeout (yields control between batches):

while True:
    for message in consumer.consume(timeout=5.0):
        handle(message)
    check_app_state()

METHOD

.close()

show/hide

Exceptions

All exceptions raised by astrocolibri derive from a single base class, so you can catch any client error with one except:

from astrocolibri.exceptions import AstrocolibriError

try:
consumer = Consumer(username="alice", password="wrong-password")
except AstrocolibriError as e:
print(f"Astro-COLIBRI error: {e}")

Exception	Inherits from	Raised when
AstrocolibriError	Exception	Base class for all errors in this library
AstrocolibriConfigError	AstrocolibriError	Invalid parameters are passed to `Consumer()`, e.g. an invalid `start_at`
AstrocolibriAuthError	AstrocolibriError	`username`/`password` are invalid or the broker refuses the connection
AstrocolibriKafkaError	AstrocolibriError	A message carries an unrecoverable Kafka error while consuming (anything other than `_PARTITION_EOF`, which is silently ignored)

Citation

Please acknowledge the use of the software and cite the Astro-COLIBRI platform papers:
- Reichherzer et al. (2021), Astro-COLIBRI — The COincidence LIBrary for Real-time Inquiry for Multimessenger Astrophysics, ApJS 256, 5, doi:10.3847/1538-4365/ac1517.
- Reichherzer et al. (2023), Astro-COLIBRI 2 — An Advanced Platform for Real-Time Multi-Messenger Discoveries, Galaxies 11, 22, doi:10.3390/galaxies11010022.

License

This source-available software is licensed under the PolyForm Noncommercial License 1.0.0. It may be used, modified, and redistributed for permitted noncommercial purposes, including use by educational institutions and public research organizations.

Commercial use requires a separate written license. Contact Astro-COLIBRI (Fabian Schüssler).

Use of Astro-COLIBRI hosted services, including the Kafka broker and its data, is governed separately by the Astro-COLIBRI Terms of Service.

Astro-COLIBRI

Astro-COLIBRI

Menu

Alert Broker & Python SDK

The Alert Broker

Getting access

Connection parameters

Available topics

What reaches the important stream

Retention and delivery

Access scope

Using a client other than the SDK

Python SDK

Installation

Quick Start

Consumer Reference

astrocolibri.Consumer(username, password, *, broker_url=None, group_id=None, start_at="earliest", security_protocol="SASL_SSL", config=None)

Description

Parameters

Raises

Examples

.subscribe(topics, *, on_assign=None, on_revoke=None, on_lost=None)

.consume(num_messages=1, timeout=-1)

Description

Parameters

Yields

Raises

Examples

.close()

Description

Context manager

Exceptions

Citation

License