124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Example HTTP MCP Client for Laravel Healthcare MCP Server
|
||
Demonstrates how to connect to MCP server via HTTP transport
|
||
"""
|
||
|
||
import json
|
||
import requests
|
||
from typing import Dict, Any, Optional
|
||
|
||
class HttpMcpClient:
|
||
def __init__(self, server_url: str = "http://localhost:3000/mcp"):
|
||
self.server_url = server_url
|
||
self.request_id = 1
|
||
self.session = requests.Session()
|
||
self.session.headers.update({
|
||
'Content-Type': 'application/json'
|
||
})
|
||
|
||
def request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""Send MCP request via HTTP"""
|
||
if params is None:
|
||
params = {}
|
||
|
||
request_data = {
|
||
"jsonrpc": "2.0",
|
||
"method": method,
|
||
"params": params,
|
||
"id": self.request_id
|
||
}
|
||
self.request_id += 1
|
||
|
||
print(f"📤 Sending MCP request: {method}")
|
||
print(f"📝 Request: {json.dumps(request_data, indent=2)}")
|
||
|
||
try:
|
||
response = self.session.post(
|
||
self.server_url,
|
||
json=request_data,
|
||
timeout=30
|
||
)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
print(f"📥 Response: {json.dumps(result, indent=2)}")
|
||
return result
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Request failed: {e}")
|
||
raise
|
||
|
||
def initialize(self) -> Dict[str, Any]:
|
||
"""Initialize MCP connection"""
|
||
return self.request('initialize', {
|
||
"protocolVersion": "2024-11-05",
|
||
"clientInfo": {
|
||
"name": "python-http-mcp-client",
|
||
"version": "1.0.0"
|
||
}
|
||
})
|
||
|
||
def list_tools(self) -> Dict[str, Any]:
|
||
"""List available tools"""
|
||
return self.request('tools/list')
|
||
|
||
def call_tool(self, tool_name: str, args: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""Execute a tool"""
|
||
if args is None:
|
||
args = {}
|
||
|
||
return self.request('tools/call', {
|
||
"name": tool_name,
|
||
"arguments": args
|
||
})
|
||
|
||
def ping(self) -> Dict[str, Any]:
|
||
"""Ping server"""
|
||
return self.request('ping')
|
||
|
||
def main():
|
||
"""Example usage"""
|
||
client = HttpMcpClient()
|
||
|
||
try:
|
||
print("🚀 Starting HTTP MCP Client...")
|
||
print("=" * 50)
|
||
|
||
# Test connection
|
||
print("1️⃣ Testing connection...")
|
||
client.initialize()
|
||
|
||
# List tools
|
||
print("\n2️⃣ Listing available tools...")
|
||
tools_response = client.list_tools()
|
||
tools = tools_response.get('result', {}).get('tools', [])
|
||
print(f"📊 Found {len(tools)} tools")
|
||
|
||
# Show first 5 tools
|
||
print("\n📋 First 5 tools:")
|
||
for i, tool in enumerate(tools[:5], 1):
|
||
print(f" {i}. {tool['name']}: {tool['description']}")
|
||
|
||
# Test tool execution (login)
|
||
print("\n3️⃣ Testing tool execution (login)...")
|
||
client.call_tool('public_manage_login', {
|
||
'email': 'test@example.com',
|
||
'password': 'password'
|
||
})
|
||
|
||
# Test ping
|
||
print("\n4️⃣ Testing ping...")
|
||
client.ping()
|
||
|
||
print("\n✅ All tests completed successfully!")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ Client test failed: {e}")
|
||
return 1
|
||
|
||
return 0
|
||
|
||
if __name__ == "__main__":
|
||
exit(main())
|