stdio

stdio client transport.

Runs an MCP server as a subprocess and exchanges newline-delimited JSON-RPC messages with it over stdin/stdout. Two pipe tasks bridge the server's pipes to the session's in-memory streams; shutdown follows the MCP spec sequence (close stdin, wait, then kill the process tree) inside a cancellation shield with every wait bounded, so a cancelled caller can neither leak a live server process nor hang on one.

get_default_environment

get_default_environment() -> dict[str, str]

Returns only the environment variables that are safe to inherit.

Source code in src/mcp/client/stdio.py

def get_default_environment() -> dict[str, str]:
    """Returns only the environment variables that are safe to inherit."""
    env: dict[str, str] = {}

    for key in DEFAULT_INHERITED_ENV_VARS:
        value = os.environ.get(key)
        if value is None:  # pragma: lax no cover
            continue

        if value.startswith("()"):  # pragma: no cover
            # Skip functions, which are a security risk
            continue  # pragma: no cover

        env[key] = value

    return env

StdioServerParameters

Bases: BaseModel

Source code in src/mcp/client/stdio.py

class StdioServerParameters(BaseModel):
    command: str
    """The executable to run to start the server."""

    args: list[str] = Field(default_factory=list)
    """Command line arguments to pass to the executable."""

    env: dict[str, str] | None = None
    """Extra environment variables, merged over get_default_environment()."""

    cwd: str | Path | None = None
    """The working directory to use when spawning the process."""

    encoding: str = "utf-8"
    """Text encoding for messages to and from the server."""

    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
    """Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers."""

command instance-attribute

command: str

The executable to run to start the server.

args class-attribute instance-attribute

args: list[str] = Field(default_factory=list)

Command line arguments to pass to the executable.

env class-attribute instance-attribute

env: dict[str, str] | None = None

Extra environment variables, merged over get_default_environment().

cwd class-attribute instance-attribute

cwd: str | Path | None = None

The working directory to use when spawning the process.

encoding class-attribute instance-attribute

encoding: str = 'utf-8'

Text encoding for messages to and from the server.

encoding_error_handler class-attribute instance-attribute

encoding_error_handler: Literal[
    "strict", "ignore", "replace"
] = "strict"

Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers.

stdio_client async

stdio_client(
    server: StdioServerParameters, errlog: TextIO = stderr
) -> AsyncGenerator[TransportStreams, None]

Spawns an MCP server subprocess and connects to it over stdin/stdout.

Raises:

Type	Description
OSError	If the server process cannot be spawned.
ValueError	If the spawn parameters are invalid (embedded NUL bytes).

Source code in src/mcp/client/stdio.py

@asynccontextmanager
async def stdio_client(
    server: StdioServerParameters, errlog: TextIO = sys.stderr
) -> AsyncGenerator[TransportStreams, None]:
    """Spawns an MCP server subprocess and connects to it over stdin/stdout.

    Raises:
        OSError: If the server process cannot be spawned.
        ValueError: If the spawn parameters are invalid (embedded NUL bytes).
    """
    command = _get_executable_command(server.command)

    process = await _create_platform_compatible_process(
        command=command,
        args=server.args,
        env=get_default_environment() | (server.env or {}),
        errlog=errlog,
        cwd=server.cwd,
    )

    # The spawn succeeded; no awaits until the task group is entered, or a
    # cancellation delivered in the gap would leak the live process.
    read_stream_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)

    shutting_down = False
    writer_done = anyio.Event()

    async def stdout_reader() -> None:
        assert process.stdout, "Opened process is missing stdout"

        stdout = TextReceiveStream(process.stdout, encoding=server.encoding, errors=server.encoding_error_handler)
        try:
            async with read_stream_writer:
                try:
                    # One line at a time; no read-ahead while a delivery is blocked.
                    buffer = ""
                    async for chunk in stdout:
                        lines = (buffer + chunk).split("\n")
                        buffer = lines.pop()
                        for line in lines:
                            try:
                                await read_stream_writer.send(_parse_line(line))
                            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
                                return  # the session is gone; only the drain below remains
                finally:
                    await _drain_stdout(process)
        except anyio.ClosedResourceError:
            pass  # our own shutdown closed the stdout stream under the read
        except (anyio.BrokenResourceError, ConnectionError):
            # Teardown noise during shutdown, a real failure otherwise; either way
            # the session sees clean closure when the read stream closes.
            if not shutting_down:
                logger.exception("Reading from the MCP server's stdout failed mid-session")

    async def stdin_writer() -> None:
        assert process.stdin, "Opened process is missing stdin"

        try:
            async with write_stream_reader:
                async for session_message in write_stream_reader:
                    json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
                    data = (json + "\n").encode(encoding=server.encoding, errors=server.encoding_error_handler)
                    await process.stdin.send(data)
        except (anyio.ClosedResourceError, anyio.BrokenResourceError, OSError):
            # The server may still be alive: close the read stream so the session
            # sees the connection end instead of a request hanging forever.
            await read_stream_writer.aclose()
        finally:
            writer_done.set()

    async def shutdown() -> None:
        """Winds the transport down: stop traffic, flush, stop the server, release the streams."""
        # Unblock the reader into its drain: a server stuck writing stdout cannot
        # read its stdin, so draining is what lets the flush below complete.
        read_stream.close()
        # Bounded window for the writer to flush already-accepted messages.
        write_stream.close()
        with anyio.move_on_after(_WRITER_FLUSH_TIMEOUT) as flush_scope:
            await writer_done.wait()
        if flush_scope.cancelled_caught:
            await anyio.lowlevel.cancel_shielded_checkpoint()  # resync coverage on 3.11 (gh-106749)
        await _stop_server_process(process)
        await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader)
        # One pass so unblocked tasks exit via their except paths before the cancel.
        await anyio.lowlevel.checkpoint()

    async with anyio.create_task_group() as tg:
        tg.start_soon(stdout_reader)
        tg.start_soon(stdin_writer)
        try:
            yield read_stream, write_stream
        finally:
            shutting_down = True
            # Shutdown must finish even under caller cancellation, or the server
            # process would leak; every wait inside is bounded. (Native
            # task.cancel() and the fallback's worker threads can still defeat it.)
            with anyio.CancelScope(shield=True):
                await shutdown()
            # Unstick pipe tasks a kill survivor's open pipe end could still block.
            tg.cancel_scope.cancel()
    # The cancel lands via throw(); one yield resyncs 3.11 coverage (gh-106749).
    await anyio.lowlevel.cancel_shielded_checkpoint()