Function send

scripts/test_mcp_interactive.py:86–94

(process: subprocess.Popen[bytes], message: dict[str, Any])

def send(process: subprocess.Popen[bytes], message: dict[str, Any]) -> None:
    if process.stdin is None:
        raise SmokeFailure("MCP stdin is unavailable")
    encoded = json.dumps(message, separators=(",", ":")).encode("utf-8") + b"\n"
    try:
        process.stdin.write(encoded)
        process.stdin.flush()
    except (BrokenPipeError, OSError) as error:
        raise SmokeFailure("MCP server closed stdin early") from error


def wait_response(