How to connect MCP server via python client?

[qsbr-liang]

Pre-submission Checklist

• I have checked that this question would not be more appropriate as an issue in a specific repository
• I have searched existing discussions and documentation for answers

Question Category

• Protocol Specification
• SDK Usage
• Server Implementation
• General Implementation
• Documentation
• Other

Your Question

I tried to write a MCP client to connect with filesystem MCP server running on my Windows.
The MCP server could be launched but the MCP client always not able to get the response from MCP server.
I don't know if my program is feasible way to connect with MCP server.
Please kindly help or advice the correct way to write a python to connect with MCP server.
import subprocess
import json
import time
import threading
from typing import Dict, Any, Optional

class MCPFileSystemClient:
def init(self, mcp_path: str, npx_path: Optional[str] = None):
self.mcp_path = mcp_path.replace("\", "/")
self.npx_path = npx_path or "npx"
self.mcp_process = None
self.response_queue = {}
self.error_messages = []
self._stop_event = threading.Event()
self.start_mcp_server()

def start_mcp_server(self):
    try:
        command = [
            "cmd", "/c", self.npx_path, "-y",
            "@modelcontextprotocol/server-filesystem", self.mcp_path
        ]
        print(f"Attempting to start MCP service: {' '.join(command)}")

        # Use binary mode (avoid text encoding issues)
        self.mcp_process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0  # No buffering
        )

        # Start thread to handle stdout (responses)
        stdout_thread = threading.Thread(target=self._process_stdout)
        stdout_thread.daemon = True
        stdout_thread.start()

        # Start thread to handle stderr
        stderr_thread = threading.Thread(target=self._process_stderr)
        stderr_thread.daemon = True
        stderr_thread.start()

        # Wait for service to start
        time.sleep(2)
        if self.mcp_process.poll() is not None:
            raise Exception(f"MCP service has exited, errors: {self.error_messages}")

        print(f"MCP filesystem service started, accessible directory: {self.mcp_path}")

    except Exception as e:
        print(f"Error starting MCP service: {e}")
        self.close()
        raise

def _process_stdout(self):
    """Process service's stdout (responses)"""
    buffer = b""
    while not self._stop_event.is_set() and self.mcp_process.poll() is None:
        try:
            # Read data chunk
            data = self.mcp_process.stdout.read(1024)
            if not data:
                time.sleep(0.1)
                continue

            buffer += data
            # Split by lines (handle both CRLF and LF)
            lines = buffer.split(b'\n')
            buffer = lines.pop()  # Last line might be incomplete, keep for next iteration

            for line in lines:
                line = line.rstrip(b'\r')  # Remove possible \r
                if line:
                    try:
                        line_str = line.decode('utf-8')
                        print(f"Received response: {line_str}")
                        response = json.loads(line_str)
                        if "id" in response:
                            self.response_queue[response["id"]] = response
                    except json.JSONDecodeError as e:
                        print(f"Invalid JSON response: {line}")
        except Exception as e:
            print(f"Error processing stdout: {e}")
            break

def _process_stderr(self):
    """Process service's stderr"""
    while not self._stop_event.is_set() and self.mcp_process.poll() is None:
        try:
            line = self.mcp_process.stderr.readline()
            if line:
                line_str = line.decode('utf-8').rstrip()
                self.error_messages.append(line_str)
                print(f"MCP service error: {line_str}")
        except Exception as e:
            print(f"Error processing stderr: {e}")
            break

def execute_command(self, command: Dict[str, Any], timeout: int = 30, retries: int = 3) -> Dict[str, Any]:
    if self.mcp_process is None:
        raise Exception("MCP service not running")

    for attempt in range(retries):
        command_id = command.setdefault(
            "id", f"cmd-{int(time.time() * 1000)}")

        # Clear old response (if any)
        if command_id in self.response_queue:
            del self.response_queue[command_id]

        try:
            # Build command (JSON + CRLF)
            command_json = json.dumps(command).encode('utf-8') + b'\r\n'
            print(
                f"Sending command (attempt {attempt+1}/{retries}): {command_json[:-2].decode('utf-8')}")

            # Send command
            self.mcp_process.stdin.write(command_json)
            self.mcp_process.stdin.flush()

            # Wait for response
            start_time = time.time()
            while time.time() - start_time < timeout:
                if command_id in self.response_queue:
                    return self.response_queue.pop(command_id)

                # Check if service crashed
                if self.mcp_process.poll() is not None:
                    raise Exception(f"MCP service has exited, errors: {self.error_messages}")

                time.sleep(0.1)

            print(f"Command {command_id} timed out, retrying...")

        except Exception as e:
            print(f"Error executing command (attempt {attempt+1}/{retries}): {e}")
            if attempt == retries - 1:
                raise

        # Wait before retry
        time.sleep(1)

    raise TimeoutError(f"Still timed out after multiple attempts ({timeout} seconds)")

def close(self):
    self._stop_event.set()
    if self.mcp_process:
        print("Closing MCP service...")
        try:
            self.mcp_process.terminate()
            self.mcp_process.wait(timeout=5)
        except Exception as e:
            print(f"Error closing service: {e}")
        self.mcp_process = None
        print("MCP service closed")

def main():
mcp_path = "D:/PythonStudy/servers"
npx_path = "npx"

try:
    client = MCPFileSystemClient(mcp_path, npx_path)

    print("\n=== Verifying service is working ===")
    info_command = {
        "name": "list_allowed_directories",
        "parameters": {}
    }
    response = client.execute_command(info_command)
    print(json.dumps(response, indent=2, ensure_ascii=False))

    print("\n=== Listing all files in current directory ===")
    search_command = {
        "name": "list_directory",
        "parameters": {
            "path": "D:/PythonStudy/servers"
        }
    }
    response = client.execute_command(search_command)
    print(json.dumps(response, indent=2, ensure_ascii=False))

except Exception as e:
    print(f"Operation failed: {e}")
    if 'client' in locals() and client:
        print("Recent service errors:")
        for msg in client.error_messages[-5:]:
            print(f"  - {msg}")
finally:
    if 'client' in locals() and client:
        client.close()

if name == "main":
main()

[Emreerdog]

Hello, it seems that you are attempting to write the MCP python client from scratch and using your custom implementation to communicate with the MCP server.

To write an MCP client in python, you should use the official MCP python SDK

https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#writing-mcp-clients

[qsbr-liang]

Thank you very much for your help. I understood now.