#!/usr/bin/env python3
"""
MCP Client helper for communicating with the MySQL MCP server via stdio.
"""
import json
import subprocess
import sys
from typing import Any
class MCPClient:
"""Simple MCP client that communicates with the server via Docker."""
def __init__(self, container_name: str = "mysql-mcp"):
self.container_name = container_name
self._request_id = 0
def _next_id(self) -> int:
self._request_id += 1
return self._request_id
def _send_request(self, method: str, params: dict[str, Any] | None = None) -> dict:
"""Send a JSON-RPC request to the MCP server."""
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": method,
}
if params:
request["params"] = params
request_json = json.dumps(request)
try:
result = subprocess.run(
[
"docker", "exec", "-i", self.container_name,
"python", "-c", f"""
import sys
import json
import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
async def main():
request = {repr(request_json)}
# For direct tool calls, we need to use the server directly
# This is a simplified approach - just echo the request for now
print(request)
asyncio.run(main())
"""
],
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
return {"error": result.stderr or "Command failed"}
return json.loads(result.stdout) if result.stdout.strip() else {}
except subprocess.TimeoutExpired:
return {"error": "Request timed out"}
except json.JSONDecodeError as e:
return {"error": f"Invalid JSON response: {e}", "raw": result.stdout}
except Exception as e:
return {"error": str(e)}
def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict:
"""Call an MCP tool."""
return self._send_request("tools/call", {
"name": tool_name,
"arguments": arguments,
})
def list_tools(self) -> dict:
"""List available tools."""
return self._send_request("tools/list")
def call_mcp_tool(tool_name: str, arguments: dict[str, Any], container: str = "mysql-mcp") -> dict:
"""
Call an MCP tool by executing a Python script inside the container.
This directly invokes the tool handlers.
"""
args_json = json.dumps(arguments)
script = f'''
import asyncio
import json
import sys
sys.path.insert(0, "/app/src")
from server import conn_manager, call_tool
async def main():
try:
result = await call_tool("{tool_name}", {args_json})
for content in result:
if hasattr(content, "text"):
print(content.text)
except Exception as e:
print(json.dumps({{"error": str(e)}}))
asyncio.run(main())
'''
try:
result = subprocess.run(
["docker", "exec", "-i", container, "python", "-c", script],
capture_output=True,
text=True,
timeout=30,
)
output = result.stdout.strip()
if result.returncode != 0:
return {"error": result.stderr or "Command failed", "output": output}
try:
return json.loads(output) if output else {}
except json.JSONDecodeError:
return {"output": output}
except subprocess.TimeoutExpired:
return {"error": "Request timed out"}
except Exception as e:
return {"error": str(e)}
if __name__ == "__main__":
# Test the client
print("Testing MCP client...")
result = call_mcp_tool("list_connections", {})
print(json.dumps(result, indent=2))
