feat: Add socket transport as an alternative to stdio transport

README.md

@@ -635,6 +635,66 @@ mcp run server.py
 
 Note that `mcp run` or `mcp dev` only supports server using FastMCP and not the low-level server variant.
 
+### Socket Transport
+
+Socket transport provides a simple and efficient communication channel between client and server, similar to stdio but without stdout pollution concerns. Unlike stdio transport which requires clean stdout for message passing, socket transport allows the server to freely use stdout for logging and other purposes.
+
+The workflow is:
+1. Client creates a TCP server and gets an available port
+2. Client starts the server process, passing the port number
+3. Server connects back to the client's TCP server
+4. Client and server exchange messages over the TCP connection
+5. When done, client closes the connection and terminates the server process
+
+This design maintains the simplicity of stdio transport while providing more flexibility for server output handling.
+
+Example server setup:
+```python
+from mcp.server.fastmcp import FastMCP
+
+# Create server with socket transport configuration
+mcp = FastMCP(
+    "SocketServer",
+    socket_host="127.0.0.1",  # Optional, defaults to 127.0.0.1
+    socket_port=3000,  # Required when using socket transport
+)
+
+# Run with socket transport
+mcp.run(transport="socket")
+```
+
+Client usage:
+```python
+from mcp.client.session import ClientSession
+from mcp.client.socket_transport import SocketServerParameters, socket_client
+
+# Create server parameters
+params = SocketServerParameters(
+    command="python",  # Server process to run
+    args=["server.py"],  # Server script and arguments
+    # Port 0 means auto-assign an available port
+    port=0,  # Optional, defaults to 0 (auto-assign)
+    host="127.0.0.1",  # Optional, defaults to 127.0.0.1
+)
+
+# Connect to server (this will start the server process)
+async with socket_client(params) as (read_stream, write_stream):
+    async with ClientSession(read_stream, write_stream) as session:
+        # Use the session...
+        await session.initialize()
+        result = await session.call_tool("echo", {"text": "Hello!"})
+```
+
+The socket transport provides:
+- Freedom to use stdout without affecting message transport
+- Standard TCP socket-based communication
+- Automatic port assignment for easy setup
+- Connection retry logic for reliability
+- Clean process lifecycle management
+- Robust error handling
+
+For a complete example, see [`examples/fastmcp/socket_example.py`](examples/fastmcp/socket_example.py).
+
 ### Streamable HTTP Transport
 
 > **Note**: Streamable HTTP transport is superseding SSE transport for production deployments.

examples/fastmcp/socket_examples/README.md

@@ -0,0 +1,79 @@
+# Socket Transport Examples
+
+This directory contains examples demonstrating the socket transport feature of FastMCP. Socket transport provides a simple and efficient communication channel between client and server, similar to stdio but without stdout pollution concerns.
+
+## Overview
+
+The socket transport works by:
+1. Client creates a TCP server and gets an available port
+2. Client starts the server process, passing the port number
+3. Server connects back to the client's TCP server
+4. Client and server exchange messages over the TCP connection
+5. When done, client closes the connection and terminates the server process
+
+## Files
+
+- `client.py` - Example client that:
+  - Creates a TCP server
+  - Starts the server process
+  - Establishes MCP session
+  - Calls example tools
+
+- `server.py` - Example server that:
+  - Connects to client's TCP server
+  - Sets up FastMCP environment
+  - Provides example tools
+  - Demonstrates logging usage
+
+## Usage
+
+1. Run with auto-assigned port (recommended):
+```bash
+python client.py
+```
+
+2. Run with specific host and port:
+```bash
+python client.py --host localhost --port 3000
+```
+
+3. Run server directly (for testing):
+```bash
+python server.py --name "Echo Server" --host localhost --port 3000 --log-level DEBUG
+```
+
+## Configuration
+
+### Client Options
+- `--host` - Host to bind to (default: 127.0.0.1)
+- `--port` - Port to use (default: 0 for auto-assign)
+
+### Server Options
+- `--name` - Server name
+- `--host` - Host to connect to
+- `--port` - Port to connect to (required)
+- `--log-level` - Logging level (DEBUG/INFO/WARNING/ERROR)
+
+## Implementation Details
+
+### Client Features
+- Automatic port assignment
+- Server process management
+- Connection retry logic
+- Error handling
+- Clean shutdown
+
+### Server Features
+- Connection retry logic
+- Custom text encoding support
+- Stdout/logging freedom
+- Error handling
+- Clean shutdown
+
+### Error Handling
+The examples demonstrate handling of:
+- Connection failures and retries
+- Invalid JSON messages
+- Text encoding errors
+- Tool execution errors
+- Process lifecycle management 

examples/fastmcp/socket_examples/client.py

@@ -0,0 +1,155 @@
+"""
+Example of using socket transport with FastMCP.
+
+This example demonstrates:
+1. Creating a FastMCP server that uses socket transport
+2. Creating a client that connects to the server using socket transport
+3. Exchanging messages between client and server
+4. Handling connection errors and retries
+5. Using custom encoding and configuration
+6. Verifying server process cleanup
+
+Usage:
+    python client.py [--host HOST] [--port PORT] [--log-level LEVEL]
+"""
+
+import argparse
+import asyncio
+import logging
+import sys
+import psutil
+from pathlib import Path
+
+from mcp.client.session import ClientSession
+from mcp.client.socket_transport import SocketServerParameters, socket_client
+from mcp.shared.exceptions import McpError
+
+# Set up logging
+logger = logging.getLogger(__name__)
+
+
+async def verify_process_cleanup(pid: int) -> bool:
+    """
+    Verify if a process with given PID exists.
+
+    Args:
+        pid: Process ID to check
+
+    Returns:
+        bool: True if process does not exist (cleaned up), False if still running
+    """
+    try:
+        process = psutil.Process(pid)
+        return False  # Process still exists
+    except psutil.NoSuchProcess:
+        return True  # Process has been cleaned up
+
+
+async def main(host: str = "127.0.0.1", port: int = 0, log_level: str = "INFO"):
+    """
+    Run the client which will start and connect to the server.
+
+    Args:
+        host: The host to use for socket communication (default: 127.0.0.1)
+        port: The port to use for socket communication (default: 0 for auto-assign)
+        log_level: Logging level (default: INFO)
+    """
+    # Configure logging
+    logging.basicConfig(
+        level=log_level,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+    server_pid = None
+    try:
+        # Create server parameters with custom configuration
+        params = SocketServerParameters(
+            # The command to run the server
+            command=sys.executable,  # Use current Python interpreter
+            # Arguments to start the server script
+            args=[
+                str(Path(__file__).parent / "server.py"),  # Updated path
+                "--name",
+                "Echo Server",
+                "--host",
+                host,
+                "--port",
+                str(port),
+                "--log-level",
+                log_level,
+            ],
+            # Socket configuration
+            host=host,
+            port=port,
+            # Optional: customize encoding (defaults shown)
+            encoding="utf-8",
+            encoding_error_handler="strict",
+        )
+
+        # Connect to server (this will start the server process)
+        async with socket_client(params) as (read_stream, write_stream):
+            # Create client session
+            async with ClientSession(read_stream, write_stream) as session:
+                try:
+                    # Initialize the session
+                    await session.initialize()
+                    logger.info("Session initialized successfully")
+
+                    # Get server process PID for verification
+                    result = await session.call_tool("get_pid_tool", {})
+                    server_pid = result.structuredContent["result"]["pid"]
+                    logger.info(f"Server process PID: {server_pid}")
+
+                    # List available tools
+                    tools = await session.list_tools()
+                    logger.info(f"Available tools: {[t.name for t in tools.tools]}")
+
+                    # Call the echo tool with different inputs
+                    messages = [
+                        "Hello from socket transport!",
+                        "Testing special chars: 世界, мир, ♥",
+                        "Testing long message: " + "x" * 1000,
+                    ]
+
+                    for msg in messages:
+                        try:
+                            result = await session.call_tool("echo_tool", {"text": msg})
+                            logger.info(f"Echo result: {result}")
+                        except McpError as e:
+                            logger.error(f"Tool call failed: {e}")
+
+                except McpError as e:
+                    logger.error(f"Session error: {e}")
+                    sys.exit(1)
+
+        # After session ends, verify server process cleanup
+        if server_pid:
+            await asyncio.sleep(0.5)  # Give some time for cleanup
+            is_cleaned = await verify_process_cleanup(server_pid)
+            if is_cleaned:
+                logger.info(
+                    f"Server process (PID: {server_pid}) was successfully cleaned up"
+                )
+            else:
+                logger.warning(f"Server process (PID: {server_pid}) is still running!")
+
+    except Exception as e:
+        logger.error(f"Connection failed: {e}", exc_info=True)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Socket transport example client")
+    parser.add_argument("--host", default="127.0.0.1", help="Host to use")
+    parser.add_argument("--port", type=int, default=0, help="Port to use (0 for auto)")
+    parser.add_argument(
+        "--log-level",
+        default="INFO",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
+        help="Logging level",
+    )
+
+    args = parser.parse_args()
+
+    # Run everything
+    asyncio.run(main(host=args.host, port=args.port, log_level=args.log_level))