"""Simple test client for the MCP server.
This is a minimal example that tests the MCP server directly without LangGraph.
Usage:
python examples/test_client.py
"""
import asyncio
import json
import sys
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def test_server():
"""Test MCP server tools."""
print("Starting MCP test client...")
print("=" * 60)
# Path to server
server_script_path = str(Path(__file__).parent.parent / "src" / "server.py")
# Server parameters
server_params = StdioServerParameters(
command=sys.executable,
args=[server_script_path],
)
# Connect to server
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize
await session.initialize()
print("Connected to MCP server\n")
# List tools
tools = await session.list_tools()
print(f"Available tools: {len(tools.tools)}")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description}")
print("\n" + "=" * 60)
print("Testing Calculator Tools")
print("=" * 60)
# Test add
result = await session.call_tool("calculator_add", {"a": 10, "b": 5})
print(f"\ncalculator_add(10, 5):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# Test divide
result = await session.call_tool("calculator_divide", {"a": 20, "b": 4})
print(f"\ncalculator_divide(20, 4):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# Test sqrt
result = await session.call_tool("calculator_sqrt", {"x": 16})
print(f"\ncalculator_sqrt(16):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# Test sin
result = await session.call_tool(
"calculator_sin", {"x": 90, "unit": "degrees"}
)
print(f"\ncalculator_sin(90, unit='degrees'):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
print("\n" + "=" * 60)
print("Testing File Operations")
print("=" * 60)
# Create a test file
result = await session.call_tool(
"file_write",
{
"file_path": "demo.txt",
"content": "Line 1\nLine 2\nLine 3\nHello World!\n",
},
)
print(f"\nfile_write(demo.txt):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# Read the file
result = await session.call_tool("file_read", {"file_path": "demo.txt"})
print(f"\nfile_read(demo.txt):")
data = json.loads(result.content[0].text)
print(f"Content:\n{data['content']}")
print(f"Metadata: {json.dumps(data['metadata'], indent=2)}")
# Read specific lines
result = await session.call_tool(
"file_read_lines", {"file_path": "demo.txt", "start": 1, "count": 2}
)
print(f"\nfile_read_lines(demo.txt, start=1, count=2):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# List files
result = await session.call_tool("file_list", {})
print(f"\nfile_list():")
print(json.dumps(json.loads(result.content[0].text), indent=2))
# Check existence
result = await session.call_tool("file_exists", {"file_path": "demo.txt"})
print(f"\nfile_exists(demo.txt):")
print(json.dumps(json.loads(result.content[0].text), indent=2))
print("\n" + "=" * 60)
print("Testing Error Handling")
print("=" * 60)
# Test division by zero
result = await session.call_tool("calculator_divide", {"a": 10, "b": 0})
print(f"\ncalculator_divide(10, 0) [should error]:")
print(result.content[0].text)
# Test file not found
result = await session.call_tool(
"file_read", {"file_path": "nonexistent.txt"}
)
print(f"\nfile_read(nonexistent.txt) [should error]:")
print(result.content[0].text)
print("\n" + "=" * 60)
print("All tests completed!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(test_server())
