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.standard_io.StandardWriter

class documentation

class StandardWriter: (source)

Constructor: StandardWriter(stream)

View In Hierarchy

Serialize bounded queued output without blocking async cancellation.

Method __init__ Start one daemon dedicated to the captured binary protocol channel.
Method _write Write whole frames and acknowledge flush completion with backpressure.
Method close Stop queued output without joining a potentially blocked OS writer.
Async Method write Wait for one complete protocol frame to reach the output stream.
Class Variable __slots__ Undocumented
Instance Variable _closed Undocumented
Instance Variable _loop Undocumented
Instance Variable _pending Undocumented
Instance Variable _queue Undocumented
Instance Variable _stream Undocumented
def __init__(self, stream: BinaryIO): (source)

Start one daemon dedicated to the captured binary protocol channel.

Parameters
stream:BinaryIOValue supplied for stream.
Returns
NoneComplete the documented operation without returning a value.
def _write(self): (source)

Write whole frames and acknowledge flush completion with backpressure.

Returns
NoneComplete the documented operation without returning a value.
def close(self): (source)

Stop queued output without joining a potentially blocked OS writer.

Returns
NoneComplete the documented operation without returning a value.
async def write(self, data: bytes): (source)

Wait for one complete protocol frame to reach the output stream.

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

Undocumented

_closed = (source)

Undocumented

_loop = (source)

Undocumented

_pending: Future[tuple[bytes, asyncio.Future[None]]] | None = (source)

Undocumented

_queue: asyncio.Queue[tuple[bytes, asyncio.Future[None]]] = (source)

Undocumented

_stream = (source)

Undocumented