Connection Errors

import shutil

# Check if command exists
command = "npx"  # or your server command
if shutil.which(command):
    print(f"✅ {command} found at: {shutil.which(command)}")
else:
    print(f"❌ {command} not found in PATH")

import asyncio
import time

async def test_server_startup():
    start_time = time.time()
    try:
        client = MCPClient.from_config_file("config.json")
        await asyncio.wait_for(client.create_all_sessions(), timeout=10)
        elapsed = time.time() - start_time
        print(f"✅ Sessions created in {elapsed:.2f}s")
    except asyncio.TimeoutError:
        elapsed = time.time() - start_time
        print(f"❌ Session creation timed out after {elapsed:.2f}s")

# Run test
await test_server_startup()

# Check file permissions
ls -la /path/to/server/executable

# Check directory permissions
ls -ld /path/to/workspace/directory

# Check current user
whoami
id

import subprocess
import json

def test_server_manually(config_file):
    with open(config_file) as f:
        config = json.load(f)

    for name, server_config in config["mcpServers"].items():
        print(f"\nTesting server: {name}")
        command = [server_config["command"]] + server_config.get("args", [])

        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=10
            )
            print(f"Return code: {result.returncode}")
            if result.stdout:
                print(f"Stdout: {result.stdout}")
            if result.stderr:
                print(f"Stderr: {result.stderr}")
        except Exception as e:
            print(f"Error: {e}")

# Test all servers
test_server_manually("config.json")

async def test_stdio_communication():
    import subprocess
    import json

    # Start server process
    process = subprocess.Popen(
        ["npx", "@playwright/mcp@latest"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )

    # Send initialization message
    init_msg = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {"capabilities": {}}
    }

    try:
        process.stdin.write(json.dumps(init_msg) + "\n")
        process.stdin.flush()

        # Read response with timeout
        import select
        ready, _, _ = select.select([process.stdout], [], [], 5)

        if ready:
            response = process.stdout.readline()
            print(f"✅ Server responded: {response}")
        else:
            print("❌ No response from server")

    finally:
        process.terminate()

await test_stdio_communication()

import aiohttp
import asyncio

async def test_http_server(url="http://localhost:8000"):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(f"{url}/health") as response:
                print(f"✅ HTTP server responding: {response.status}")
                text = await response.text()
                print(f"Response: {text}")
        except Exception as e:
            print(f"❌ HTTP connection failed: {e}")

await test_http_server()

import websockets
import asyncio

async def test_websocket(uri="ws://localhost:8000/ws"):
    try:
        async with websockets.connect(uri) as websocket:
            print("✅ WebSocket connected")

            # Send ping
            await websocket.send("ping")
            response = await asyncio.wait_for(websocket.recv(), timeout=5)
            print(f"Response: {response}")

    except Exception as e:
        print(f"❌ WebSocket failed: {e}")

await test_websocket()

# Check what's using port 8000
lsof -i :8000
netstat -tulpn | grep 8000

# Find available ports
python -c "import socket; s=socket.socket(); s.bind(('',0)); print(s.getsockname()[1]); s.close()"

# Check firewall status
sudo ufw status
sudo iptables -L

# Test connection
telnet localhost 8000
nc -zv localhost 8000

# Open port in firewall
sudo ufw allow 8000
sudo iptables -A INPUT -p tcp --dport 8000 -j ACCEPT

# Or disable firewall for testing
sudo ufw disable

# Install Node.js and npm
RUN apt-get update && apt-get install -y nodejs npm && \
    apt-get clean && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY pyproject.toml ./

# Install in editable mode
COPY . .
RUN pip install -e ".[dev,anthropic,openai]"

# Expose necessary ports
EXPOSE 8000

# Set proper networking
CMD ["python", "main.py", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  mcp-use:
    build: .
    ports:
      - "8000:8000"

# Mount with proper permissions
docker run -v /host/workspace:/container/workspace:rw myapp

# Check mounted volume permissions
docker exec -it container ls -la /container/workspace

import logging

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Enable specific loggers
logging.getLogger("mcp_use").setLevel(logging.DEBUG)
logging.getLogger("asyncio").setLevel(logging.INFO)

# Create client with debug mode
client = MCPClient.from_config_file("config.json", debug=True)

# Monitor network traffic
sudo tcpdump -i lo port 8000 -A

# Monitor with netstat
watch -n 1 'netstat -tulpn | grep 8000'

# Use curl for HTTP testing
curl -v http://localhost:8000/health

import psutil
import subprocess

def monitor_server_process(command_name):
    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
        try:
            if command_name in proc.info['name'] or \
               any(command_name in arg for arg in proc.info['cmdline']):
                print(f"Found process: PID={proc.info['pid']}")
                print(f"Command: {' '.join(proc.info['cmdline'])}")
                print(f"Status: {proc.status()}")
                print(f"Memory: {proc.memory_info().rss / 1024 / 1024:.1f}MB")
                return proc.info['pid']
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass

    print(f"No process found for: {command_name}")
    return None

# Monitor specific server
pid = monitor_server_process("npx")

import asyncio
from typing import Optional

class ResilientMCPClient:
    def __init__(self, config_file: str, max_retries: int = 3):
        self.config_file = config_file
        self.max_retries = max_retries
        self._client: Optional[MCPClient] = None

    async def connect_with_retry(self):
        for attempt in range(self.max_retries):
            try:
                self._client = MCPClient.from_config_file(self.config_file)
                await self._client.create_all_sessions()
                print(f"✅ Connected on attempt {attempt + 1}")
                return self._client
            except Exception as e:
                print(f"❌ Attempt {attempt + 1} failed: {e}")
                if attempt < self.max_retries - 1:
                    wait_time = 2 ** attempt  # Exponential backoff
                    print(f"Retrying in {wait_time}s...")
                    await asyncio.sleep(wait_time)
                else:
                    raise

    async def ensure_connected(self):
        if not self._client:
            return await self.connect_with_retry()

        # Test connection
        try:
            # Test if sessions are active
            active_sessions = self._client.get_all_active_sessions()
            if len(active_sessions) > 0:
                return self._client
        except:
            print("Connection lost, reconnecting...")
            return await self.connect_with_retry()

# Usage
resilient_client = ResilientMCPClient("config.json", max_retries=3)
client = await resilient_client.ensure_connected()

import asyncio
from datetime import datetime, timedelta

class ServerHealthMonitor:
    def __init__(self, client: MCPClient, check_interval: int = 30):
        self.client = client
        self.check_interval = check_interval
        self.last_check = datetime.now()
        self.is_healthy = True

    async def health_check(self):
        try:
            # Check if we have active sessions
            active_sessions = self.client.get_all_active_sessions()
            self.is_healthy = len(active_sessions) > 0
            self.last_check = datetime.now()
            return self.is_healthy
        except Exception as e:
            print(f"Health check failed: {e}")
            self.is_healthy = False
            return False

    async def start_monitoring(self):
        while True:
            await self.health_check()
            if not self.is_healthy:
                print("⚠️ Server unhealthy, attempting reconnection...")
                try:
                    await self.client.close_all_sessions()
                    await self.client.create_all_sessions()
                    await self.health_check()
                except Exception as e:
                    print(f"Reconnection failed: {e}")

            await asyncio.sleep(self.check_interval)

# Usage
monitor = ServerHealthMonitor(client, check_interval=30)
asyncio.create_task(monitor.start_monitoring())