feat: expose errlog on stdio transport (#1991)

Co-authored-by: William Easton <williamseaston@gmail.com>
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>

src/fastmcp/client/transports.py

@@ -8,7 +8,7 @@ import sys
 import warnings
 from collections.abc import AsyncIterator
 from pathlib import Path
-from typing import Any, Literal, TypeVar, cast, overload
+from typing import Any, Literal, TextIO, TypeVar, cast, overload
 
 import anyio
 import httpx
@@ -313,6 +313,7 @@ class StdioTransport(ClientTransport):
         env: dict[str, str] | None = None,
         cwd: str | None = None,
         keep_alive: bool | None = None,
+        log_file: Path | TextIO | None = None,
     ):
         """
         Initialize a Stdio transport.
@@ -326,6 +327,11 @@ class StdioTransport(ClientTransport):
                        Defaults to True. When True, the subprocess remains active
                        after the connection context exits, allowing reuse in
                        subsequent connections.
+            log_file: Optional path or file-like object where subprocess stderr will
+                   be written. Can be a Path or TextIO object. Defaults to sys.stderr
+                   if not provided. When a Path is provided, the file will be created
+                   if it doesn't exist, or appended to if it does. When set, server
+                   errors will be written to this file instead of appearing in the console.
         """
         self.command = command
         self.args = args
@@ -334,6 +340,7 @@ class StdioTransport(ClientTransport):
         if keep_alive is None:
             keep_alive = True
         self.keep_alive = keep_alive
+        self.log_file = log_file
 
         self._session: ClientSession | None = None
         self._connect_task: asyncio.Task | None = None
@@ -368,6 +375,7 @@ class StdioTransport(ClientTransport):
                 args=self.args,
                 env=self.env,
                 cwd=self.cwd,
+                log_file=self.log_file,
                 session_kwargs=session_kwargs,
                 ready_event=self._ready_event,
                 stop_event=self._stop_event,
@@ -421,6 +429,7 @@ async def _stdio_transport_connect_task(
     args: list[str],
     env: dict[str, str] | None,
     cwd: str | None,
+    log_file: Path | TextIO | None,
     session_kwargs: SessionKwargs,
     ready_event: anyio.Event,
     stop_event: anyio.Event,
@@ -438,7 +447,19 @@ async def _stdio_transport_connect_task(
                     env=env,
                     cwd=cwd,
                 )
-                transport = await stack.enter_async_context(stdio_client(server_params))
+                # Handle log_file: Path needs to be opened, TextIO used as-is
+                if log_file is None:
+                    log_file_handle = sys.stderr
+                elif isinstance(log_file, Path):
+                    log_file_handle = open(log_file, "a")
+                    stack.callback(log_file_handle.close)
+                else:
+                    # Must be TextIO - use it directly
+                    log_file_handle = log_file
+
+                transport = await stack.enter_async_context(
+                    stdio_client(server_params, errlog=log_file_handle)
+                )
                 read_stream, write_stream = transport
                 session_future.set_result(
                     await stack.enter_async_context(
@@ -471,6 +492,7 @@ class PythonStdioTransport(StdioTransport):
         cwd: str | None = None,
         python_cmd: str = sys.executable,
         keep_alive: bool | None = None,
+        log_file: Path | TextIO | None = None,
     ):
         """
         Initialize a Python transport.
@@ -485,6 +507,11 @@ class PythonStdioTransport(StdioTransport):
                        Defaults to True. When True, the subprocess remains active
                        after the connection context exits, allowing reuse in
                        subsequent connections.
+            log_file: Optional path or file-like object where subprocess stderr will
+                   be written. Can be a Path or TextIO object. Defaults to sys.stderr
+                   if not provided. When a Path is provided, the file will be created
+                   if it doesn't exist, or appended to if it does. When set, server
+                   errors will be written to this file instead of appearing in the console.
         """
         script_path = Path(script_path).resolve()
         if not script_path.is_file():
@@ -502,6 +529,7 @@ class PythonStdioTransport(StdioTransport):
             env=env,
             cwd=cwd,
             keep_alive=keep_alive,
+            log_file=log_file,
         )
         self.script_path = script_path
 
@@ -516,6 +544,7 @@ class FastMCPStdioTransport(StdioTransport):
         env: dict[str, str] | None = None,
         cwd: str | None = None,
         keep_alive: bool | None = None,
+        log_file: Path | TextIO | None = None,
     ):
         script_path = Path(script_path).resolve()
         if not script_path.is_file():
@@ -529,6 +558,7 @@ class FastMCPStdioTransport(StdioTransport):
             env=env,
             cwd=cwd,
             keep_alive=keep_alive,
+            log_file=log_file,
         )
         self.script_path = script_path
 
@@ -544,6 +574,7 @@ class NodeStdioTransport(StdioTransport):
         cwd: str | None = None,
         node_cmd: str = "node",
         keep_alive: bool | None = None,
+        log_file: Path | TextIO | None = None,
     ):
         """
         Initialize a Node transport.
@@ -558,6 +589,11 @@ class NodeStdioTransport(StdioTransport):
                        Defaults to True. When True, the subprocess remains active
                        after the connection context exits, allowing reuse in
                        subsequent connections.
+            log_file: Optional path or file-like object where subprocess stderr will
+                   be written. Can be a Path or TextIO object. Defaults to sys.stderr
+                   if not provided. When a Path is provided, the file will be created
+                   if it doesn't exist, or appended to if it does. When set, server
+                   errors will be written to this file instead of appearing in the console.
         """
         script_path = Path(script_path).resolve()
         if not script_path.is_file():
@@ -570,7 +606,12 @@ class NodeStdioTransport(StdioTransport):
             full_args.extend(args)
 
         super().__init__(
-            command=node_cmd, args=full_args, env=env, cwd=cwd, keep_alive=keep_alive
+            command=node_cmd,
+            args=full_args,
+            env=env,
+            cwd=cwd,
+            keep_alive=keep_alive,
+            log_file=log_file,
         )
         self.script_path = script_path
 

tests/client/test_stdio.py

@@ -253,3 +253,112 @@ class TestKeepAlive:
         with pytest.raises(RuntimeError, match="Client failed to connect"):
             async with client:
                 pass
+
+
+class TestLogFile:
+    @pytest.fixture
+    def stdio_script_with_stderr(self, tmp_path):
+        script = inspect.cleandoc('''
+            import sys
+            from fastmcp import FastMCP
+
+            mcp = FastMCP()
+
+            @mcp.tool
+            def write_error(message: str) -> str:
+                """Writes a message to stderr and returns it"""
+                print(message, file=sys.stderr, flush=True)
+                return message
+
+            if __name__ == "__main__":
+                mcp.run()
+            ''')
+        script_file = tmp_path / "stderr_script.py"
+        script_file.write_text(script)
+        return script_file
+
+    async def test_log_file_parameter_accepted_by_stdio_transport(self, tmp_path):
+        """Test that log_file parameter can be set on StdioTransport"""
+        log_file_path = tmp_path / "errors.log"
+        transport = StdioTransport(
+            command="python", args=["script.py"], log_file=log_file_path
+        )
+        assert transport.log_file == log_file_path
+
+    async def test_log_file_parameter_accepted_by_python_stdio_transport(
+        self, tmp_path, stdio_script_with_stderr
+    ):
+        """Test that log_file parameter can be set on PythonStdioTransport"""
+        log_file_path = tmp_path / "errors.log"
+        transport = PythonStdioTransport(
+            script_path=stdio_script_with_stderr, log_file=log_file_path
+        )
+        assert transport.log_file == log_file_path
+
+    async def test_log_file_parameter_accepts_textio(self, tmp_path):
+        """Test that log_file parameter can accept a TextIO object"""
+        log_file_path = tmp_path / "errors.log"
+        with open(log_file_path, "w") as log_file:
+            transport = StdioTransport(
+                command="python", args=["script.py"], log_file=log_file
+            )
+            assert transport.log_file == log_file
+
+    async def test_log_file_captures_stderr_output_with_path(
+        self, tmp_path, stdio_script_with_stderr
+    ):
+        """Test that stderr output is written to the log_file when using Path"""
+        log_file_path = tmp_path / "errors.log"
+
+        transport = PythonStdioTransport(
+            script_path=stdio_script_with_stderr, log_file=log_file_path
+        )
+        client = Client(transport=transport)
+
+        async with client:
+            await client.call_tool("write_error", {"message": "Test error message"})
+
+        # Need to wait a bit for stderr to flush
+        await asyncio.sleep(0.1)
+
+        content = log_file_path.read_text()
+        assert "Test error message" in content
+
+    async def test_log_file_captures_stderr_output_with_textio(
+        self, tmp_path, stdio_script_with_stderr
+    ):
+        """Test that stderr output is written to the log_file when using TextIO"""
+        log_file_path = tmp_path / "errors.log"
+
+        with open(log_file_path, "w") as log_file:
+            transport = PythonStdioTransport(
+                script_path=stdio_script_with_stderr, log_file=log_file
+            )
+            client = Client(transport=transport)
+
+            async with client:
+                await client.call_tool(
+                    "write_error", {"message": "Test error with TextIO"}
+                )
+
+            # Need to wait a bit for stderr to flush
+            await asyncio.sleep(0.1)
+
+        content = log_file_path.read_text()
+        assert "Test error with TextIO" in content
+
+    async def test_log_file_none_uses_default_behavior(
+        self, tmp_path, stdio_script_with_stderr
+    ):
+        """Test that log_file=None uses default stderr handling"""
+        transport = PythonStdioTransport(
+            script_path=stdio_script_with_stderr, log_file=None
+        )
+        client = Client(transport=transport)
+
+        async with client:
+            # Should work without error even without explicit log_file
+            result = await client.call_tool(
+                "write_error", {"message": "Default stderr"}
+            )
+            assert result.data == "Default stderr"