emergent-client

Python SDK for building event-driven workflows on the Emergent engine. Connect to a running engine over Unix IPC and publish or consume messages through three typed primitives: Source, Handler, and Sink.

from emergent import EmergentSink

async for msg in EmergentSink.messages("my_sink", ["timer.tick"]):
    print(msg.payload)

Install

pip install emergent-client

Or with uv:

uv add emergent-client

Then import:

from emergent import EmergentSource, EmergentHandler, EmergentSink

Three Primitives

Every Emergent workflow is composed of Sources, Handlers, and Sinks. Each primitive has a single, well-defined role:

Quick Start

Sink -- consume messages

A Sink subscribes to message types and processes each one as it arrives. EmergentSink.messages is a convenience method that connects, subscribes, and yields messages in a single call:

from emergent import EmergentSink

async for msg in EmergentSink.messages("my_sink", ["timer.tick"]):
    data = msg.payload_as(dict)
    print(f"Tick #{data['sequence']}")

For explicit lifecycle control, connect and subscribe separately:

async with await EmergentSink.connect("my_sink") as sink:
    async with await sink.subscribe(["timer.tick", "timer.filtered"]) as stream:
        async for msg in stream:
            print(msg.message_type, msg.payload)

Source -- publish messages

A Source publishes messages into the engine. It cannot subscribe:

from emergent import EmergentSource

async with await EmergentSource.connect("my_source") as source:
    await source.publish("sensor.reading", {"value": 42.5, "unit": "celsius"})

Handler -- subscribe and publish

A Handler subscribes to incoming messages and publishes new ones. Use caused_by to link output messages to the input that triggered them:

from emergent import EmergentHandler, create_message

async with await EmergentHandler.connect("order_processor") as handler:
    async with await handler.subscribe(["order.created"]) as stream:
        async for msg in stream:
            await handler.publish(
                create_message("order.processed")
                .caused_by(msg.id)
                .payload({"status": "ok"})
            )

Publishing Messages

Every primitive that can publish supports three calling styles. All three produce the same result:

# Shorthand -- type string and payload dict
await source.publish("timer.tick", {"count": 1})

# MessageBuilder -- fluent API with auto-build
await source.publish(
    create_message("timer.tick").payload({"count": 1})
)

# Full EmergentMessage object
await source.publish(message)

Streaming Publish

Publish a collection or async stream of messages. Each message is sent individually so subscribers begin consuming immediately. Both methods return the count of successfully published messages and stop on the first error.

# From a list or any Iterable
messages = [
    create_message("record.imported").payload(record)
    for record in records
]
count = await source.publish_all(messages)

# From an async generator or any AsyncIterable
async def generate_messages():
    for i in range(100):
        yield create_message("batch.item").payload({"index": i})

count = await source.publish_stream(generate_messages())

Both publish_all and publish_stream are available on EmergentSource and EmergentHandler.

Building Messages

create_message returns a fluent builder for constructing immutable EmergentMessage instances:

from emergent import create_message

msg = (
    create_message("sensor.reading")
    .payload({"value": 42.5, "unit": "celsius"})
    .metadata({"sensor_id": "temp-01", "location": "room-a"})
    .build()
)

Link messages into traceable chains with caused_by and correlated_with:

reply = (
    create_message("order.confirmed")
    .caused_by(original_msg.id)
    .correlated_with(request_id)
    .payload({"confirmed": True})
    .build()
)

The builder sets id (TypeID format) and timestamp_ms automatically. Call .build() explicitly when you need the message object, or pass the builder directly to publish(), which calls .build() for you.

Subscribing to Messages

subscribe accepts a list or variadic arguments:

# List form
stream = await sink.subscribe(["timer.tick", "timer.filtered"])

# Variadic form
stream = await sink.subscribe("timer.tick", "timer.filtered")

Iterate over the returned MessageStream with async for:

async for msg in stream:
    data = msg.payload_as(dict)
    print(data["count"])

MessageStream implements AsyncIterator and the async context manager protocol, so you can use async with for automatic cleanup:

async with await sink.subscribe(["timer.tick"]) as stream:
    async for msg in stream:
        print(msg.payload)

The stream also ends when the connection to the engine is lost, so the loop stops and the code after it runs. See Error Handling.

A client has one live stream, so call subscribe once with every topic. From 0.14.0 a second subscribe on the same client ends the earlier stream, so an async for over it stops, and returns a new one. The engine keeps the earlier subscriptions, so the new stream receives the earlier topics as well as the new ones. The earlier stream ends when the second call starts, even if that call then fails. On 0.13.1 and earlier the earlier stream was left open and unfed, and nothing ever ended it, not even close(). To read two sets of topics apart from each other, connect two clients.

Typed payloads with Pydantic

payload_as validates dict payloads against Pydantic models automatically:

from pydantic import BaseModel

class SensorReading(BaseModel):
    value: float
    unit: str

async for msg in EmergentSink.messages("my_sink", ["sensor.reading"]):
    reading = msg.payload_as(SensorReading)
    print(f"{reading.value} {reading.unit}")

Resource Cleanup

All primitives implement the async context manager protocol. Use async with for automatic cleanup (recommended), or call close() / disconnect() manually:

# Automatic cleanup (recommended)
async with await EmergentSink.connect("my_sink") as sink:
    ...

# Manual cleanup
sink = await EmergentSink.connect("my_sink")
# ... use sink ...
await sink.disconnect()

The SDK subscribes to system.shutdown internally. When the Emergent engine signals a graceful shutdown, active message streams close automatically.

Helper Functions

run_source, run_handler, and run_sink eliminate connection and signal-handling boilerplate. Each helper connects, sets up SIGTERM/SIGINT handlers, runs your callback, and disconnects on completion:

import asyncio
from emergent import run_source, run_handler, run_sink, create_message

# Source -- custom event loop with shutdown signal
async def timer_logic(source, shutdown_event):
    count = 0
    while not shutdown_event.is_set():
        try:
            await asyncio.wait_for(shutdown_event.wait(), timeout=3.0)
            break
        except asyncio.TimeoutError:
            count += 1
            await source.publish(
                create_message("timer.tick").payload({"count": count})
            )

await run_source("my_timer", timer_logic)

# Handler -- called once per message
async def process(msg, handler):
    await handler.publish(
        create_message("processed").caused_by(msg.id).payload({"done": True})
    )

await run_handler("my_handler", ["raw.event"], process)

# Sink -- called once per message
async def consume(msg):
    print(msg.payload)

await run_sink("my_sink", ["timer.tick"], consume)

The name argument is optional. When omitted or set to None, the helper reads from the EMERGENT_NAME environment variable.

run_source sets shutdown_event on SIGTERM, on SIGINT, and, from 0.14.0, when the engine closes the connection. A Source subscribes to nothing, so no stream ends to tell it the engine is gone, and a lost connection is the only notice an engine that was killed ever gives. On 0.13.1 and earlier only the two signals set the event, so a Source that caught its publish errors kept running against a dead socket unless the engine had spawned it on Linux, where the kernel signals the child when its parent dies. run_handler and run_sink end with their message stream, which a lost connection closes.

Error Handling

All SDK errors extend EmergentError and include a machine-readable code property. Catch specific error types for precise control:

from emergent import (
    EmergentSource,
    SocketNotFoundError,
    ConnectionError,
    TimeoutError,
)

try:
    source = await EmergentSource.connect("my_source")
except SocketNotFoundError as e:
    print(f"Engine not running at: {e.socket_path}")
except TimeoutError as e:
    print(f"Timed out after {e.timeout}s")
except ConnectionError as e:
    print(f"Connection failed: {e}")

Error Types

SubscriptionError and DiscoveryError extend ConnectionError. On SDK release 0.13.1 and earlier they were exported and never raised: each of those failures raised a plain ConnectionError with the code CONNECTION_FAILED. An except ConnectionError still catches them, so put the specific class first. Code that compares code against CONNECTION_FAILED for one of these failures now sees the code in the table. PublishError extends EmergentError directly, as it always has.

When the socket refuses a write, which is what happens once the engine has gone away, the SDK error keeps the operating system's error as __cause__. From 0.14.0 a refused publish() or publish_ack() raises PublishError, and a refused subscribe(), discover() or other request raises ConnectionError. On 0.13.1 and earlier all of them raised the built-in error itself (ConnectionResetError, BrokenPipeError), which is not an EmergentError, so an except EmergentError never saw it. The built-in ConnectionError and the SDK's share a name and nothing else: code that caught OSError for these calls should now catch the SDK class and read __cause__.

When the engine closes the connection, every call still waiting for an answer (publish_ack(), discover(), subscribe(), get_topology(), get_my_subscriptions()) raises ConnectionError("Connection closed") at once, and the message stream ends, so an async for over it stops. That holds from 0.14.0. On 0.13.1 and earlier each call waited out its own timeout, 30 seconds by default, and then raised TimeoutError. The stream ended only when the read failed: when the engine closed the connection cleanly it never ended, so an async for consumer waited until the process was stopped.

A frame the engine sends with a malformed body is never raised to the caller. From 0.14.0 it is logged and skipped, and the subscription stays open. On 0.13.1 and earlier one such frame ended the read loop and closed the stream.

A message whose envelope carries a field this SDK does not know is delivered with that field ignored, from 0.14.0. On 0.13.1 and earlier the wire model forbade unknown fields, so one new envelope field from the engine would have stopped every Python subscriber from receiving messages.

Message Shape

Every message flowing through Emergent follows the same envelope:

Use msg.payload_as(MyModel) to validate and convert the payload to a typed Pydantic model or any other type.

System Events

The Emergent engine broadcasts lifecycle events that your primitives can subscribe to:

Use the typed payload classes for safe access:

from emergent import SystemEventPayload

if msg.message_type.startswith("system.started."):
    event = msg.payload_as(SystemEventPayload)
    print(f"{event.name} ({event.kind}) started with PID {event.pid}")

if msg.message_type.startswith("system.error."):
    event = msg.payload_as(SystemEventPayload)
    if event.is_error():
        print(f"{event.name} failed: {event.error}")

Requirements

• Python 3.12 or later
• A running Emergent engine with the EMERGENT_SOCKET environment variable set

License

MIT OR Apache-2.0