#!/usr/bin/env python3
import asyncio
import json
import subprocess
import sys
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
async def test_mcp_server():
# Start the server as a subprocess
server_process = subprocess.Popen(
[sys.executable, "main.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Connect to the server
async with stdio_client(server_process) as streams:
async with ClientSession(streams[0], streams[1]) as session:
# Initialize the session3
await session.initialize()
# List available tools
tools = await session.list_tools()
print("Available tools:")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description}")
# Test the echo tool
echo_result = await session.call_tool("echo", {"text": "Hello, MCP!"})
print(f"\nEcho result: {echo_result.content[0].text}")
# Test the LLM prediction tool
llm_result = await session.call_tool("llm_predict", {
"prompt": "What is the capital of France?",
"max_tokens": 50
})
print(f"\nLLM result: {llm_result.content[0].text}")
except Exception as e:
print(f"Error: {e}")
finally:
server_process.terminate()
server_process.wait()
if __name__ == "__main__":
asyncio.run(test_mcp_server())
