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.StandardReader

class documentation

class StandardReader: (source)

Constructor: StandardReader(stream, limit)

View In Hierarchy

Read at most one queued frame using an interruptible async queue boundary.

Method __init__ Start one daemon so an idle Windows pipe cannot delay executor shutdown.
Method _offer Apply queue backpressure in the daemon rather than allocating callbacks.
Method _read Read bounded chunks and drain oversized frames without accumulating them.
Method close Stop future deliveries without waiting on an uninterruptible OS read.
Async Method readline Return the next frame or propagate a real input transport failure.
Class Variable __slots__ Undocumented
Instance Variable _closed Undocumented
Instance Variable _limit Undocumented
Instance Variable _loop Undocumented
Instance Variable _pending Undocumented
Instance Variable _queue Undocumented
Instance Variable _stream Undocumented
def __init__(self, stream: BinaryIO, limit: int): (source)

Start one daemon so an idle Windows pipe cannot delay executor shutdown.

Parameters
stream:BinaryIOValue supplied for stream.
limit:intValue supplied for limit.
Returns
NoneComplete the documented operation without returning a value.
def _offer(self, item: bytes | OSError) -> bool: (source)

Apply queue backpressure in the daemon rather than allocating callbacks.

Parameters
item:bytes | OSErrorValue supplied for item.
Returns
boolResult of the operation described above.
def _read(self): (source)

Read bounded chunks and drain oversized frames without accumulating them.

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

Stop future deliveries without waiting on an uninterruptible OS read.

Returns
NoneComplete the documented operation without returning a value.
async def readline(self) -> bytes: (source)

Return the next frame or propagate a real input transport failure.

Returns
bytesThe next frame or propagate a real input transport failure.
__slots__: tuple[str, ...] = (source)

Undocumented

_closed = (source)

Undocumented

_limit = (source)

Undocumented

_loop = (source)

Undocumented

_pending: Future[None] | None = (source)

Undocumented

_queue: asyncio.Queue[bytes | OSError] = (source)

Undocumented

_stream = (source)

Undocumented