Function _close_subprocess_transport

src/mcp/client/stdio.py:304–317

Closes the asyncio subprocess transport, if there is one. The transport otherwise stays open (and warns at GC) while a surviving descendant holds a pipe end; nothing public exposes it, hence the attribute walk. No-op on trio and the Windows fallback.

(process: ServerProcess)

def _close_subprocess_transport(process: ServerProcess) -> None:
    """Closes the asyncio subprocess transport, if there is one.

    The transport otherwise stays open (and warns at GC) while a surviving
    descendant holds a pipe end; nothing public exposes it, hence the attribute
    walk. No-op on trio and the Windows fallback.
    """
    transport = getattr(getattr(process, "_process", None), "_transport", None)
    # Duck-typed: uvloop's UVProcessTransport is not an asyncio.SubprocessTransport.
    close = getattr(transport, "close", None)
    if callable(close):
        # close() on <=3.12 can raise PermissionError re-killing a setuid child.
        with suppress(PermissionError):
            close()


def _get_executable_command(command: str) -> str: