test_mcp.py•4.12 kB
#!/usr/bin/env python3
"""
Test script for MCP Server
Tests the math operations MCP server functionality using subprocess
"""
import asyncio
import json
import sys
from contextlib import asynccontextmanager
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
@asynccontextmanager
async def get_mcp_client():
"""Create an MCP client connection to the server"""
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"],
env=None
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
async def test_mcp_server():
"""
Test the MCP server by listing tools and calling them
"""
print("Starting MCP Server Tests...\n")
try:
async with get_mcp_client() as session:
print("[OK] Connected to MCP server successfully!\n")
# Test 1: List available tools
print("=" * 50)
print("TEST 1: Listing Available Tools")
print("=" * 50)
tools = await session.list_tools()
print(f"Found {len(tools.tools)} tools:\n")
for tool in tools.tools:
print(f"[Tool] {tool.name}")
print(f" Description: {tool.description}")
print(f" Input Schema: {json.dumps(tool.inputSchema, indent=2)}\n")
# Test 2: Call the 'add' tool
print("=" * 50)
print("TEST 2: Testing 'add' Tool")
print("=" * 50)
add_params = {"a": 15.5, "b": 24.5}
print(f"Input: {add_params}")
add_result = await session.call_tool("add", add_params)
print(f"[OK] Result:")
for content in add_result.content:
print(f" {content.text}\n")
# Test 3: Call the 'subtract' tool
print("=" * 50)
print("TEST 3: Testing 'subtract' Tool")
print("=" * 50)
subtract_params = {"a": 100, "b": 42}
print(f"Input: {subtract_params}")
subtract_result = await session.call_tool("subtract", subtract_params)
print(f"[OK] Result:")
for content in subtract_result.content:
print(f" {content.text}\n")
# Test 4: Test with negative numbers
print("=" * 50)
print("TEST 4: Testing with Negative Numbers")
print("=" * 50)
negative_params = {"a": -10, "b": -5}
print(f"Add Input: {negative_params}")
negative_add_result = await session.call_tool("add", negative_params)
print(f"[OK] Result:")
for content in negative_add_result.content:
print(f" {content.text}\n")
# Test 5: Test with decimal precision
print("=" * 50)
print("TEST 5: Testing Decimal Precision")
print("=" * 50)
decimal_params = {"a": 0.1, "b": 0.2}
print(f"Add Input: {decimal_params}")
decimal_result = await session.call_tool("add", decimal_params)
print(f"[OK] Result:")
for content in decimal_result.content:
print(f" {content.text}\n")
# Test 6: Test error handling
print("=" * 50)
print("TEST 6: Testing Error Handling")
print("=" * 50)
try:
print("Calling non-existent tool 'multiply'...")
await session.call_tool("multiply", {"a": 5, "b": 3})
except Exception as e:
print(f"[ERROR] Expected error caught: {type(e).__name__}: {e}\n")
print("=" * 50)
print("[SUCCESS] All Tests Completed!")
print("=" * 50)
except Exception as e:
print(f"\n[FAILED] Test failed with error: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
asyncio.run(test_mcp_server())