test_client.py

#!/usr/bin/env python3
"""
Simple MCP Client to test the DeepSeek MCP Server

This script demonstrates how to interact with your running MCP server.
"""

import asyncio
import json
import sys
import subprocess
from pathlib import Path

async def test_mcp_server():
    """Test the MCP server by sending direct requests"""
    
    print("Testing DeepSeek MCP Server...")
    
    # Test 1: List available tools
    print("\n1. Testing tool listing...")
    list_tools_request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/list"
    }
    
    try:
        # Start server process for testing
        server_process = subprocess.Popen(
            [sys.executable, "server.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            cwd=Path(__file__).parent
        )
        
        # Send request
        request_json = json.dumps(list_tools_request) + "\n"
        stdout, stderr = server_process.communicate(input=request_json, timeout=30)
        
        if stdout:
            response = json.loads(stdout.strip())
            print("Available tools:")
            for tool in response.get("result", {}).get("tools", []):
                print(f"   - {tool['name']}: {tool['description']}")
        
        server_process.terminate()
        
    except Exception as e:
        print(f"Error testing server: {e}")
        if 'server_process' in locals():
            server_process.terminate()
        return False
    
    return True

def test_chat_directly():
    """Test chat functionality directly with the MCP interface"""
    print("\n2. Testing chat functionality directly...")
    
    try:
        from mcp_interface import MCPLLMInterface
        from config import config
        
        # Create interface
        interface = MCPLLMInterface(config.model_path)
        
        # Test chat
        async def run_chat_test():
            # Load model
            await interface._load_model()
            print("Model loaded successfully")
            
            # Test chat
            chat_args = {
                "message": "Hello! Please respond with just 'Hello from DeepSeek!'",
                "max_tokens": 50,
                "temperature": 0.3
            }
            
            response = await interface._handle_chat(chat_args)
            print(f"Chat response: {response[0].text}")
            
            return True
        
        return asyncio.run(run_chat_test())
        
    except Exception as e:
        print(f"Error in direct chat test: {e}")
        return False

def main():
    """Main test function"""
    print("DeepSeek MCP Server Test Client")
    print("=" * 50)
    
    # Test 1: Direct functionality test
    success = test_chat_directly()
    
    if success:
        print("\nServer is working correctly!")
        print("\nHow to use with Claude Desktop:")
        print("1. Make sure Claude Desktop is installed")
        print("2. Restart Claude Desktop")
        print("3. Your DeepSeek server should appear in available tools")
        print("4. Use: @deepseek-mcp-server [your message]")
        
        print("\nServer Configuration:")
        try:
            from config import config
            print(f"   Model: {Path(config.model_path).name}")
            print(f"   GPU Layers: {config.n_gpu_layers}")
            print(f"   Context Size: {config.n_ctx}")
            print(f"   Max Tokens: {config.default_max_tokens}")
        except:
            print("   Configuration loaded successfully")
            
    else:
        print("\nServer test failed. Please check the setup.")
        
    print("\n" + "=" * 50)

if __name__ == "__main__":
    main()