orionis.mcp.manager.McpManager

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

class documentation

class McpManager(IMcpManager): (source)

Constructor: McpManager(app, router, config, bus)

View In Hierarchy

Own one compiled registry shared by native HTTP, STDIO, DI and the facade.

Method	`__init__`	Capture application services without opening transports or performing I/O.
Method	`_compile`	Validate and compile a server exactly once per application manager.
Async Method	`_publish`	Scope events to a declaration already registered in this application.
Async Method	`dispatchHttp`	Use the same server and invoker reached by its local registration.
Method	`finalizeRoutes`	Index final fluent paths once at boot, preserving constant-time dispatch.
Method	`getLocalServer`	Return compiled local metadata by its configured handle.
Method	`getWebServer`	Return compiled HTTP metadata by normalized native route path.
Method	`local`	Register an explicit handle without accepting dynamic module imports.
Async Method	`promptsChanged`	Publish a prompt catalog change to the matching server's listeners.
Async Method	`resourcesChanged`	Publish a resource catalog change to the matching server's listeners.
Async Method	`resourceUpdated`	Publish one changed URI without resolving its content or identity.
Method	`servers`	Return an immutable transport/address/declaration inventory.
Async Method	`shutdown`	Wake subscriptions so their request owners can complete and clean up.
Async Method	`startLocal`	Serve only a configured local handle using native standard streams.
Async Method	`toolsChanged`	Publish a catalog change without retaining the current identity.
Method	`web`	Register one native POST endpoint with API authentication semantics.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_app`	Undocumented
Instance Variable	`_bus`	Undocumented
Instance Variable	`_compiled`	Undocumented
Instance Variable	`_config`	Undocumented
Instance Variable	`_local`	Undocumented
Instance Variable	`_router`	Undocumented
Instance Variable	`_web`	Undocumented

def __init__(self, app: IApplication, router: IRouter, config: McpConfig, bus: IMcpEventBus): (source) ¶

Capture application services without opening transports or performing I/O.

Parameters
app:`IApplication`	Application container supplying configuration and dependencies.
router:`IRouter`	Value supplied for `router`.
config:`McpConfig`	Validated configuration controlling this component.
bus:`IMcpEventBus`	Value supplied for `bus`.

Returns
`None`	Complete the documented operation without returning a value.

def _compile(self, server: type[Server]) -> CompiledMcpServer: (source) ¶

Validate and compile a server exactly once per application manager.

Parameters
server:`type[Server]`	Value supplied for `server`.

Returns
`CompiledMcpServer`	Result of the operation described above.

async def _publish(self, server: type[Server], method: str, uri: str | None = None): (source) ¶

Scope events to a declaration already registered in this application.

Parameters
server:`type[Server]`	Value supplied for `server`.
method:`str`	Value supplied for `method`.
uri:`str \| None`	Value supplied for `uri`.

Returns
`None`	Complete the documented operation without returning a value.

async def dispatchHttp(self, request: Request) -> Response: (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.dispatchHttp

Use the same server and invoker reached by its local registration.

Parameters
request:`Request`	Current request and its trusted execution context.

Returns
`Response`	Result of the operation described above.

def finalizeRoutes(self): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.finalizeRoutes

Index final fluent paths once at boot, preserving constant-time dispatch.

Returns
`None`	Complete the documented operation without returning a value.

def getLocalServer(self, name: str) -> CompiledMcpServer: (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.getLocalServer

Return compiled local metadata by its configured handle.

Parameters
name:`str`	Value supplied for `name`.

Returns
`CompiledMcpServer`	Compiled local metadata by its configured handle.

def getWebServer(self, path: str) -> CompiledMcpServer: (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.getWebServer

Return compiled HTTP metadata by normalized native route path.

Parameters
path:`str`	Value supplied for `path`.

Returns
`CompiledMcpServer`	Compiled HTTP metadata by normalized native route path.

def local(self, name: str, server: type[Server]): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.local

Parameters
name:`str`	Value supplied for `name`.
server:`type[Server]`	Value supplied for `server`.

Returns
`None`	Complete the documented operation without returning a value.

async def promptsChanged(self, server: type[Server]): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.promptsChanged

Publish a prompt catalog change to the matching server's listeners.

Parameters
server:`type[Server]`	Value supplied for `server`.

Returns
`None`	Complete the documented operation without returning a value.

async def resourcesChanged(self, server: type[Server]): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.resourcesChanged

Publish a resource catalog change to the matching server's listeners.

Parameters
server:`type[Server]`	Value supplied for `server`.

Returns
`None`	Complete the documented operation without returning a value.

async def resourceUpdated(self, server: type[Server], uri: str): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.resourceUpdated

Publish one changed URI without resolving its content or identity.

Parameters
server:`type[Server]`	Value supplied for `server`.
uri:`str`	Value supplied for `uri`.

Returns
`None`	Complete the documented operation without returning a value.

def servers(self) -> tuple[tuple[str, str, type[Server]], ...]: (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.servers

Return an immutable transport/address/declaration inventory.

Returns
`tuple[tuple[str, str, type[Server]], ...]`	An immutable transport/address/declaration inventory.

async def shutdown(self): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.shutdown

Wake subscriptions so their request owners can complete and clean up.

Returns
`None`	Complete the documented operation without returning a value.

async def startLocal(self, name: str): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.startLocal

Serve only a configured local handle using native standard streams.

Parameters
name:`str`	Value supplied for `name`.

Returns
`None`	Complete the documented operation without returning a value.

async def toolsChanged(self, server: type[Server]): (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.toolsChanged

Publish a catalog change without retaining the current identity.

Parameters
server:`type[Server]`	Value supplied for `server`.

Returns
`None`	Complete the documented operation without returning a value.

def web(self, path: str, server: type[Server]) -> FluentRoute: (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.web

Parameters
path:`str`	Value supplied for `path`.
server:`type[Server]`	Value supplied for `server`.

Returns
`FluentRoute`	Result of the operation described above.

__slots__: tuple[str, ...] = (source) ¶

overrides orionis.mcp.contracts.manager.IMcpManager.__slots__

Undocumented

_app = (source) ¶

Undocumented

_bus = (source) ¶

Undocumented

_compiled: dict[type[Server], CompiledMcpServer] = (source) ¶

Undocumented

_config = (source) ¶

Undocumented

_local: dict[str, CompiledMcpServer] = (source) ¶

Undocumented

_router = (source) ¶

Undocumented

_web: dict[str, tuple[FluentRoute, CompiledMcpServer, McpHttpTransport]] = (source) ¶

Undocumented