mcp_client_http.py•4.72 kB
#!/usr/bin/env python3
"""
HTTP MCP Client for testing remote MCP servers
"""
import json
import requests
import asyncio
from typing import Dict, Any, Optional
class HTTPMCPClient:
"""HTTP client for MCP servers"""
def __init__(self, base_url: str):
"""Initialize HTTP MCP client"""
self.base_url = base_url.rstrip('/')
self.mcp_endpoint = f"{self.base_url}/mcp"
self.session = requests.Session()
self.request_id = 1
def _make_request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make JSON-RPC request to MCP server"""
payload = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method
}
if params:
payload["params"] = params
self.request_id += 1
try:
response = self.session.post(
self.mcp_endpoint,
json=payload,
headers={"Content-Type": "application/json"},
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {
"jsonrpc": "2.0",
"id": payload["id"],
"error": {
"code": -32000,
"message": f"HTTP request failed: {str(e)}"
}
}
def initialize(self) -> Dict[str, Any]:
"""Initialize MCP connection"""
return self._make_request("initialize")
def list_tools(self) -> Dict[str, Any]:
"""Get list of available tools"""
return self._make_request("tools/list")
def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Call a specific tool"""
params = {
"name": tool_name,
"arguments": arguments
}
return self._make_request("tools/call", params)
def health_check(self) -> Dict[str, Any]:
"""Check server health"""
try:
response = self.session.get(f"{self.base_url}/health", timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": f"Health check failed: {str(e)}"}
def main():
"""Test the HTTP MCP client"""
import sys
if len(sys.argv) < 2:
print("Usage: python mcp_client_http.py <base_url>")
print("Example: python mcp_client_http.py http://localhost:8000")
sys.exit(1)
base_url = sys.argv[1]
client = HTTPMCPClient(base_url)
print(f"🔗 Testing MCP server at: {base_url}")
# Test health check
print("\n1. Health Check:")
health = client.health_check()
print(f" Status: {health}")
# Initialize MCP
print("\n2. Initialize MCP:")
init_result = client.initialize()
if "error" in init_result:
print(f" Error: {init_result['error']}")
sys.exit(1)
else:
print(f" Success: {init_result.get('result', {}).get('serverInfo', {}).get('name', 'Unknown')}")
# List tools
print("\n3. List Tools:")
tools_result = client.list_tools()
if "error" in tools_result:
print(f" Error: {tools_result['error']}")
else:
tools = tools_result.get('result', {}).get('tools', [])
print(f" Available tools: {len(tools)}")
for tool in tools:
print(f" - {tool.get('name')}: {tool.get('description')}")
# Test query_resume tool
print("\n4. Test query_resume tool:")
query_result = client.call_tool("query_resume", {
"query": "What programming languages and technologies do I know?"
})
if "error" in query_result:
print(f" Error: {query_result['error']}")
else:
content = query_result.get('result', {}).get('content', [])
if content and len(content) > 0:
response_text = content[0].get('text', '')
print(f" Response: {response_text[:200]}...")
else:
print(" No content returned")
# Test get_agent_info tool
print("\n5. Test get_agent_info tool:")
info_result = client.call_tool("get_agent_info", {})
if "error" in info_result:
print(f" Error: {info_result['error']}")
else:
content = info_result.get('result', {}).get('content', [])
if content and len(content) > 0:
info_text = content[0].get('text', '')
print(f" Info: {info_text[:200]}...")
else:
print(" No content returned")
print("\n✅ MCP HTTP client test completed!")
if __name__ == "__main__":
main()