#!/usr/bin/env python3
"""
Test script for MCP server functionality.
This script tests the MCP server by creating a client connection
and exercising the available tools.
"""
import asyncio
import json
import sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def test_mcp_server():
"""Test the MCP server functionality."""
# Start the server process
server_params = StdioServerParameters(
command="python",
args=["server.py"],
)
print("🚀 Starting MCP server test...")
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the client
await session.initialize()
print("✅ Server initialized successfully")
# Test 1: List available tools
print("\n📋 Testing tool listing...")
tools = await session.list_tools()
print(f"Available tools: {[tool.name for tool in tools.tools]}")
expected_tools = {"echo", "get_memory", "store_memory", "llm_chat"}
actual_tools = {tool.name for tool in tools.tools}
if expected_tools.issubset(actual_tools):
print("✅ All expected tools are available")
else:
missing = expected_tools - actual_tools
print(f"❌ Missing tools: {missing}")
return False
# Test 2: Echo tool
print("\n🔄 Testing echo tool...")
echo_result = await session.call_tool("echo", {"text": "Hello, MCP!"})
echo_content = echo_result.content[0].text
print(f"Echo response: {echo_content}")
if echo_content == "Hello, MCP!":
print("✅ Echo tool works correctly")
else:
print(f"❌ Echo tool failed. Expected 'Hello, MCP!', got '{echo_content}'")
return False
# Test 3: Memory storage and retrieval
print("\n💾 Testing memory system...")
# Store a memory
store_result = await session.call_tool("store_memory", {
"conversation_id": "test_conv_123",
"content": "This is a test memory",
"metadata": {"test": True}
})
store_response = store_result.content[0].text
print(f"Store memory response: {store_response}")
if "successfully" in store_response.lower():
print("✅ Memory storage works")
else:
print("❌ Memory storage failed")
return False
# Retrieve the memory
get_result = await session.call_tool("get_memory", {
"conversation_id": "test_conv_123"
})
get_response = get_result.content[0].text
print(f"Retrieved memory: {get_response[:100]}...")
if "test memory" in get_response.lower():
print("✅ Memory retrieval works")
else:
print("❌ Memory retrieval failed")
return False
# Test 4: LLM chat (only if API keys are available)
print("\n🤖 Testing LLM integration...")
try:
llm_result = await session.call_tool("llm_chat", {
"message": "Say 'Hello from MCP test' and nothing else",
"model": "gpt-3.5-turbo"
})
llm_response = llm_result.content[0].text
print(f"LLM response: {llm_response}")
if "error" not in llm_response.lower():
print("✅ LLM integration works")
else:
print("⚠️ LLM integration needs API key configuration")
except Exception as e:
print(f"⚠️ LLM test skipped (likely missing API key): {e}")
print("\n🎉 All tests completed successfully!")
return True
async def test_server_startup():
"""Test that the server starts up correctly."""
print("🔧 Testing server startup...")
try:
from server import MCPServer
server = MCPServer()
print("✅ Server instantiation successful")
return True
except Exception as e:
print(f"❌ Server startup failed: {e}")
return False
def main():
"""Run all tests."""
print("🧪 MCP Server Test Suite")
print("=" * 40)
# Test 1: Server startup
if not asyncio.run(test_server_startup()):
sys.exit(1)
# Test 2: Full server functionality
try:
if not asyncio.run(test_mcp_server()):
sys.exit(1)
except KeyboardInterrupt:
print("\n⏹️ Tests interrupted by user")
sys.exit(1)
except Exception as e:
print(f"\n❌ Test suite failed with error: {e}")
sys.exit(1)
print("\n✨ All tests passed! Your MCP server is working correctly.")
if __name__ == "__main__":
main()
