File size: 3,754 Bytes
ba18ff2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
#!/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()
|