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.transport.stdio.McpStdioTransport

class documentation

class McpStdioTransport: (source)

Constructor: McpStdioTransport(app, compiled, config, bus)

View In Hierarchy

Keep the reader available while bounded request tasks execute independently.

Method __init__ Prepare request-local task ownership without touching process streams.
Async Method _accept Stop queued output without joining a potentially blocked OS writer.
Method _cancel Ignore malformed, unknown and already completed cancellation messages.
Method _done Remove completed task ownership and notify the reader of I/O failures.
Async Method _invoke Execute one call in its own DI scope, sanitizing only application errors.
Async Method _readLoop Race each cancellable read against a failed request's transport I/O.
Async Method _stream Preserve backpressure and close an owned producer on all exits.
Async Method _write Serialize frames and suppress output after EOF or owner shutdown.
Async Method run Run until EOF, joining all owned request cleanup before returning.
Async Method runStandardStreams Bind portable daemon-backed streams and keep application prints on stderr.
Class Variable __slots__ Undocumented
Instance Variable _active Undocumented
Instance Variable _app Undocumented
Instance Variable _closing Undocumented
Instance Variable _config Undocumented
Instance Variable _dispatcher Undocumented
Instance Variable _failed Undocumented
Instance Variable _failure Undocumented
Instance Variable _lock Undocumented
Instance Variable _running Undocumented
def __init__(self, app: IApplication, compiled: CompiledMcpServer, config: McpConfig, bus: IMcpEventBus): (source)

Prepare request-local task ownership without touching process streams.

Parameters
app:IApplicationApplication container supplying configuration and dependencies.
compiled:CompiledMcpServerPrecompiled metadata shared by request executions.
config:McpConfigValidated configuration controlling this component.
bus:IMcpEventBusValue supplied for bus.
Returns
NoneComplete the documented operation without returning a value.
async def _accept(self, data: bytes, writer: LineWriter): (source)

Stop queued output without joining a potentially blocked OS writer.

Parameters
data:bytesValue supplied for data.
writer:LineWriterOutput stream receiving complete messages with backpressure.
Returns
NoneComplete the documented operation without returning a value.
def _cancel(self, request: JsonRpcRequest): (source)

Ignore malformed, unknown and already completed cancellation messages.

Parameters
request:JsonRpcRequestCurrent request and its trusted execution context.
Returns
NoneComplete the documented operation without returning a value.
def _done(self, request_id: str | int, task: asyncio.Task[None]): (source)

Remove completed task ownership and notify the reader of I/O failures.

Parameters
request_id:str | intValue supplied for request_id.
task:asyncio.Task[None]Value supplied for task.
Returns
NoneComplete the documented operation without returning a value.
async def _invoke(self, request: JsonRpcRequest, writer: LineWriter): (source)

Execute one call in its own DI scope, sanitizing only application errors.

Parameters
request:JsonRpcRequestCurrent request and its trusted execution context.
writer:LineWriterOutput stream receiving complete messages with backpressure.
Returns
NoneComplete the documented operation without returning a value.
async def _readLoop(self, reader: LineReader, writer: LineWriter): (source)

Race each cancellable read against a failed request's transport I/O.

Parameters
reader:LineReaderInput stream supplying bounded messages.
writer:LineWriterOutput stream receiving complete messages with backpressure.
Returns
NoneComplete the documented operation without returning a value.
async def _stream(self, writer: LineWriter, source: AsyncIterator[bytes]): (source)

Preserve backpressure and close an owned producer on all exits.

Parameters
writer:LineWriterOutput stream receiving complete messages with backpressure.
source:AsyncIterator[bytes]Source whose values or lifecycle are consumed by this operation.
Returns
NoneComplete the documented operation without returning a value.
async def _write(self, writer: LineWriter, data: bytes): (source)

Serialize frames and suppress output after EOF or owner shutdown.

Parameters
writer:LineWriterOutput stream receiving complete messages with backpressure.
data:bytesValue supplied for data.
Returns
NoneComplete the documented operation without returning a value.
async def run(self, reader: LineReader, writer: LineWriter): (source)

Run until EOF, joining all owned request cleanup before returning.

Parameters
reader:LineReaderInput stream supplying bounded messages.
writer:LineWriterOutput stream receiving complete messages with backpressure.
Returns
NoneComplete the documented operation without returning a value.
async def runStandardStreams(self): (source)

Bind portable daemon-backed streams and keep application prints on stderr.

Returns
NoneComplete the documented operation without returning a value.
__slots__: tuple[str, ...] = (source)

Undocumented

_active: dict[str | int, asyncio.Task[None]] = (source)

Undocumented

_app = (source)

Undocumented

_closing: bool = (source)

Undocumented

_config = (source)

Undocumented

_dispatcher = (source)

Undocumented

_failed = (source)

Undocumented

_failure: BaseException | None = (source)

Undocumented

_lock = (source)

Undocumented

_running: bool = (source)

Undocumented