diff --git a/gateway/mcp/mcp.py b/gateway/mcp/mcp.py index 73dadce..4c55a8c 100644 --- a/gateway/mcp/mcp.py +++ b/gateway/mcp/mcp.py @@ -6,6 +6,7 @@ import json import os import select import asyncio +import platform from invariant_sdk.async_client import AsyncClient from invariant_sdk.types.append_messages import AppendMessagesRequest @@ -527,6 +528,49 @@ async def process_line( mcp_process.stdin.flush() +async def wait_for_stdin_input(loop: asyncio.AbstractEventLoop, stdin_fd: int) -> tuple[bytes | None, str]: + """ + Platform-specific implementation to wait for and read input from stdin. + + Args: + loop: The asyncio event loop + stdin_fd: The file descriptor for stdin + + Returns: + tuple[bytes | None, str]: A tuple containing: + - The data read from stdin or None + - Status: 'eof' if EOF detected, 'data' if data available, 'wait' if no data yet + """ + if platform.system() == "Windows": + # On Windows, we can't use select for stdin + # Instead, we'll use a brief sleep and then try to read + await asyncio.sleep(0.01) + try: + chunk = await loop.run_in_executor(None, lambda: os.read(stdin_fd, 4096)) + if not chunk: # Empty bytes means EOF + return None, 'eof' + return chunk, 'data' + except (BlockingIOError, OSError): + # No data available yet + return None, 'wait' + else: + # On Unix-like systems, use select + ready, _, _ = await loop.run_in_executor( + None, lambda: select.select([stdin_fd], [], [], 0.1) + ) + + if not ready: + # No input available, yield to other tasks + await asyncio.sleep(0.01) + return None, 'wait' + + # Read available data + chunk = await loop.run_in_executor(None, lambda: os.read(stdin_fd, 4096)) + if not chunk: # Empty bytes means EOF + return None, 'eof' + return chunk, 'data' + + async def run_stdio_input_loop( ctx: McpContext, mcp_process: subprocess.Popen, @@ -543,30 +587,26 @@ async def run_stdio_input_loop( try: while True: - # Check for input using select - ready, _, _ = await loop.run_in_executor( - None, lambda: select.select([stdin_fd], [], [], 0.1) - ) - - if not ready: - # No input available, yield to other tasks - await asyncio.sleep(0.01) + # Get input using platform-specific method + chunk, status = await wait_for_stdin_input(loop, stdin_fd) + + if status == 'eof': + # EOF detected, break the loop + break + elif status == 'wait': + # No data available yet, continue polling continue + elif status == 'data': + # We got some data, process it + buffer += chunk - # Read available data - chunk = await loop.run_in_executor(None, lambda: os.read(stdin_fd, 4096)) - if not chunk: - break # EOF + # Process complete lines + while b"\n" in buffer: + line, buffer = buffer.split(b"\n", 1) + if not line: + continue - buffer += chunk - - # Process complete lines - while b"\n" in buffer: - line, buffer = buffer.split(b"\n", 1) - if not line: - continue - - await process_line(ctx, mcp_process, line) + await process_line(ctx, mcp_process, line) except (BrokenPipeError, KeyboardInterrupt): # Broken pipe = client disappeared, just start shutdown mcp_log("Client disconnected or keyboard interrupt") diff --git a/uv.lock b/uv.lock index 0ad4015..b0e8f1d 100644 --- a/uv.lock +++ b/uv.lock @@ -249,7 +249,7 @@ wheels = [ [[package]] name = "invariant-gateway" -version = "0.0.4" +version = "0.0.5" source = { editable = "." } dependencies = [ { name = "fastapi" },