enterprise_ai.runtime.bus
Source: sdks/python/src/enterprise_ai/runtime/bus.py. Back to the Python SDK reference index.
This page is generated by tools/docs/gen_sdk_reference.py from the SDK sources and the OpenAPI text folders. Do not edit it by hand: change the source and regenerate.
Model BusMessage
One message on the bus.
Fields:
| Name | Type | Required | Description |
|---|---|---|---|
content | str | yes | Text of the message. |
session_id | str | None | no | Session the message belongs to, or None. |
channel | str | no | Channel the conversation comes from. |
sender | str | no | Author of the message (default user). |
metadata | dict[str, Any] | no | Free key-value pairs stored with the object and returned as they are; the platform never interprets them. |
BusMessage.from_json
from_json(data: bytes) -> BusMessage
Rebuild a message from its JSON encoding (the reverse of to_json).
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
data | bytes | yes | JSON bytes holding the message fields. |
Returns: BusMessage. A BusMessage: One message on the bus.
BusMessage.to_json
to_json() -> bytes
Encode every field of the message as JSON.
This element takes no parameter.
Returns: bytes. UTF-8 JSON bytes; values that are not JSON types are written as strings.
Class MessageBus
In-process asyncio bus with an inbound queue, an outbound queue and subscribers.
Attributes:
| Name | Type | Required | Description |
|---|---|---|---|
inbound | asyncio.Queue[BusMessage] | yes | asyncio.Queue of the messages from the channels to the agent loop. |
outbound | asyncio.Queue[BusMessage] | yes | asyncio.Queue of the messages from the agent loop to the channels. |
Constructor MessageBus
MessageBus() -> None
Create a bus with empty inbound and outbound queues and no subscriber.
This element takes no parameter.
MessageBus.consume_inbound
async consume_inbound(timeout: float | None = None) -> BusMessage | None
Next inbound message, or None after timeout seconds.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
timeout | float | None | None | no | Maximum wait in seconds; None waits until a message arrives. |
Returns: BusMessage \| None. A BusMessage: One message on the bus.
MessageBus.get_inbound_nowait
get_inbound_nowait() -> BusMessage | None
Take the next inbound message without waiting.
This element takes no parameter.
Returns: BusMessage \| None. A BusMessage: One message on the bus.
MessageBus.publish_inbound
async publish_inbound(message: BusMessage) -> None
Put a message in the inbound queue, then call the inbound subscribers.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
message | BusMessage | yes | Message sent by a channel to the agent. |
Returns: None.
MessageBus.publish_outbound
async publish_outbound(message: BusMessage) -> None
Put a message in the outbound queue, then call the outbound subscribers.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
message | BusMessage | yes | Message sent by the agent to the channels. |
Returns: None.
MessageBus.subscribe
subscribe(topic: str, handler: Handler) -> Callable[[], None]
Call handler for every message of topic (inbound or outbound). Returns an unsubscribe.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
topic | str | yes | Topic to follow: inbound or outbound. | |
handler | Handler | yes | Function called with each BusMessage; it may be a coroutine function, which is awaited. |
Returns: Callable[[], None]. Function without argument that removes the subscription.
Class NatsMessageBus
Same API backed by NATS subjects <prefix>.inbound and <prefix>.outbound.
MessageBus
Attributes:
| Name | Type | Required | Description |
|---|---|---|---|
prefix | yes | Prefix of the NATS subjects. | |
url | yes | Address of the NATS server. |
Constructor NatsMessageBus
NatsMessageBus(url: str = 'nats://127.0.0.1:4222', prefix: str = 'eai.agent') -> None
Create a NATS backed bus; needs the nats extra (nats-py), otherwise raises ImportError. Call connect() before use.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
url | str | 'nats://127.0.0.1:4222' | no | Address of the NATS server; defaults to a local server. |
prefix | str | 'eai.agent' | no | Prefix of the NATS subjects (default eai.agent). |
Raises: ImportError.
NatsMessageBus.close
async close() -> None
Drain and close the NATS connection when it is open.
This element takes no parameter.
Returns: None.
NatsMessageBus.connect
async connect() -> NatsMessageBus
Connect to NATS and relay the messages of <prefix>.inbound into the local inbound queue.
This element takes no parameter.
Returns: NatsMessageBus. A NatsMessageBus: Same API backed by NATS subjects <prefix>.inbound and <prefix>.outbound.
NatsMessageBus.publish_inbound
async publish_inbound(message: BusMessage) -> None
Publish a message on the <prefix>.inbound subject; it reaches the queue through the relay.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
message | BusMessage | yes | Message sent by a channel to the agent. |
Returns: None.
NatsMessageBus.publish_outbound
async publish_outbound(message: BusMessage) -> None
Publish a message on the <prefix>.outbound subject, then call the local outbound subscribers.
| Name | Type | Default | Required | Description |
|---|---|---|---|---|
message | BusMessage | yes | Message sent by the agent to the channels. |
Returns: None.