ReferenceAPI referenceagentspartykernelconsole

StreamConsole

Console backed by text streams, defaulting to the process stdio.

Streams are resolved on every call rather than captured in __init__: sys.stdout is routinely replaced after construction (test capture, output redirection), and a captured handle would silently ignore that.

Functions

func__init__(self, reader=None, writer=None) -> None

Create a console over reader/writer, defaulting to stdio.

› Source code
def __init__(
    self,
    reader: TextIO | None = None,
    writer: TextIO | None = None,
) -> None:
    """Create a console over *reader*/*writer*, defaulting to stdio.

    Args:
        reader: Input stream; ``sys.stdin`` when omitted.
        writer: Output stream; ``sys.stdout`` when omitted.
    """
    self._reader = reader
    self._writer = writer
paramreaderTextIO | None
= None

Input stream; sys.stdin when omitted.

paramwriterTextIO | None
= None

Output stream; sys.stdout when omitted.

Returns

None
funcask(self, prompt) -> str

Write the prompt, then read one line without blocking the loop.

› Source code
async def ask(self, prompt: str) -> str:
    """Write the prompt, then read one line without blocking the loop."""
    out = self._out
    out.write(prompt)
    out.flush()
    return await asyncio.to_thread(self._read_line)
parampromptstr

Returns

str
funcshow(self, text) -> None

Write text and a newline to the output stream.

› Source code
def show(self, text: str) -> None:
    """Write *text* and a newline to the output stream.

    Args:
        text: The line to print.
    """
    out = self._out
    out.write(f'{text}\n')
    out.flush()
paramtextstr

The line to print.

Returns

None

On this page