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.

class documentation

class IMcpManager(ABC): (source)

Known subclasses: orionis.mcp.manager.McpManager

View In Hierarchy

Share compiled servers, native routes, local transports, and change events.

Async Method dispatchHttp Handle a request through the shared compiled server's HTTP transport.
Method finalizeRoutes Index final native route paths after fluent prefixes and groups.
Method getLocalServer Find the compiled server assigned to a local handle.
Method getWebServer Find the compiled server assigned to an HTTP path.
Method local Register a server for an explicit local CLI handle.
Async Method promptsChanged Publish that an authorized prompt catalog may have changed.
Async Method resourcesChanged Publish that an authorized resource catalog may have changed.
Async Method resourceUpdated Publish a resource change to listeners that requested that URI.
Method servers Describe registered transports without exposing mutable registries.
Async Method shutdown Wake active subscription streams for graceful completion.
Async Method startLocal Serve one registered local server until EOF or cancellation.
Async Method toolsChanged Publish that an authorized tool catalog may have changed.
Method web Register a server on one native HTTP POST route.
Class Variable __slots__ Undocumented
async def dispatchHttp(self, request: Request) -> Response: (source)

Handle a request through the shared compiled server's HTTP transport.

Parameters
request:RequestCurrent request and its trusted execution context.
Returns
ResponseResult of the operation described above.
def finalizeRoutes(self): (source)

Index final native route paths after fluent prefixes and groups.

Returns
NoneComplete the documented operation without returning a value.
def getLocalServer(self, name: str) -> CompiledMcpServer: (source)

Find the compiled server assigned to a local handle.

Parameters
name:strValue supplied for name.
Returns
CompiledMcpServerResult of the operation described above.
def getWebServer(self, path: str) -> CompiledMcpServer: (source)

Find the compiled server assigned to an HTTP path.

Parameters
path:strValue supplied for path.
Returns
CompiledMcpServerResult of the operation described above.
def local(self, name: str, server: type[Server]): (source)

Register a server for an explicit local CLI handle.

Parameters
name:strValue supplied for name.
server:type[Server]Value supplied for server.
Returns
NoneComplete the documented operation without returning a value.
async def promptsChanged(self, server: type[Server]): (source)

Publish that an authorized prompt catalog may have changed.

Parameters
server:type[Server]Value supplied for server.
Returns
NoneComplete the documented operation without returning a value.
async def resourcesChanged(self, server: type[Server]): (source)

Publish that an authorized resource catalog may have changed.

Parameters
server:type[Server]Value supplied for server.
Returns
NoneComplete the documented operation without returning a value.
async def resourceUpdated(self, server: type[Server], uri: str): (source)

Publish a resource change to listeners that requested that URI.

Parameters
server:type[Server]Value supplied for server.
uri:strValue supplied for uri.
Returns
NoneComplete the documented operation without returning a value.
def servers(self) -> tuple[tuple[str, str, type[Server]], ...]: (source)

Describe registered transports without exposing mutable registries.

Returns
tuple[tuple[str, str, type[Server]], ...]Result of the operation described above.
async def shutdown(self): (source)

Wake active subscription streams for graceful completion.

Returns
NoneComplete the documented operation without returning a value.
async def startLocal(self, name: str): (source)

Serve one registered local server until EOF or cancellation.

Parameters
name:strValue supplied for name.
Returns
NoneComplete the documented operation without returning a value.
async def toolsChanged(self, server: type[Server]): (source)

Publish that an authorized tool catalog may have changed.

Parameters
server:type[Server]Value supplied for server.
Returns
NoneComplete the documented operation without returning a value.
def web(self, path: str, server: type[Server]) -> FluentRoute: (source)

Register a server on one native HTTP POST route.

Parameters
path:strValue supplied for path.
server:type[Server]Value supplied for server.
Returns
FluentRouteResult of the operation described above.
__slots__: tuple = (source)

Undocumented