ORIONIS API REFERENCE

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

orionis.mcp.contracts.event_bus.IMcpEventBus

class documentation

class IMcpEventBus(ABC): (source)

Known subclasses: orionis.mcp.subscriptions.event_bus.InMemoryMcpEventBus

View In Hierarchy

Publish server changes without retaining identities or requests.

Method listen Open a bounded subscription, releasing it on iterator close.
Async Method publish Publish a notification kind to the corresponding server.
Async Method shutdown Wake subscribers for a graceful end.
Class Variable __slots__ Undocumented
def listen(self, server: type, filters: SubscriptionFilter) -> AsyncIterator[tuple[str, str | None]]: (source)
overridden in orionis.mcp.subscriptions.event_bus.InMemoryMcpEventBus

Open a bounded subscription, releasing it on iterator close.

Parameters
server:typeValue supplied for server.
filters:SubscriptionFilterValue supplied for filters.
Returns
AsyncIterator[tuple[str, str | None]]Result of the operation described above.
async def publish(self, server: type, method: str, uri: str | None = None): (source)
overridden in orionis.mcp.subscriptions.event_bus.InMemoryMcpEventBus

Publish a notification kind to the corresponding server.

Parameters
server:typeValue supplied for server.
method:strValue supplied for method.
uri:str | NoneValue supplied for uri.
Returns
NoneComplete the documented operation without returning a value.
async def shutdown(self): (source)
overridden in orionis.mcp.subscriptions.event_bus.InMemoryMcpEventBus

Wake subscribers for a graceful end.

Returns
NoneComplete the documented operation without returning a value.
__slots__: tuple = (source)
overridden in orionis.mcp.subscriptions.event_bus.InMemoryMcpEventBus

Undocumented