File size: 5,408 Bytes
b0f9887 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | #!/usr/bin/env python3
"""
ECH0-PRIME MCP Server Demo Client
Demonstrates how to interact with the ECH0-PRIME MCP server
"""
import requests
import json
import time
from typing import Dict, Any, List
class ECH0MCPClient:
"""Client for interacting with the ECH0-PRIME MCP Server"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
self.session = requests.Session()
def health_check(self) -> Dict[str, Any]:
"""Check if the MCP server is running"""
try:
response = self.session.get(f"{self.base_url}/health")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": f"Health check failed: {e}"}
def list_tools(self) -> List[Dict[str, Any]]:
"""Get list of available tools"""
try:
response = self.session.get(f"{self.base_url}/tools")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return [{"error": f"Failed to list tools: {e}"}]
def call_tool(self, tool_name: str, **kwargs) -> Dict[str, Any]:
"""Call a specific tool with arguments"""
try:
payload = {"tool": tool_name, "args": kwargs}
response = self.session.post(
f"{self.base_url}/tools/call",
json=payload,
timeout=300 # 5 minutes for complex operations
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": f"Tool call failed: {e}"}
def demonstrate_capabilities(self):
"""Demonstrate various MCP server capabilities"""
print("🧠 ECH0-PRIME MCP Server Demonstration")
print("=" * 50)
# Health check
print("\n🏥 Health Check:")
health = self.health_check()
print(json.dumps(health, indent=2))
# List tools
print("\n🛠️ Available Tools:")
tools = self.list_tools()
for tool in tools:
if "error" not in tool:
print(f" • {tool['name']}: {tool['description'][:60]}...")
# Demonstrate cognitive tools
print("\n🧠 Testing Cognitive Tools:")
# Test ArXiv scanning
print("\n📚 Testing ArXiv Scanner:")
arxiv_result = self.call_tool("scan_arxiv", query="consciousness", max_results=2)
print(json.dumps(arxiv_result, indent=2))
# Test memory operations
print("\n💾 Testing Memory System:")
memory_result = self.call_tool("store_memory",
key="demo_fact",
value="ECH0-PRIME demonstrates measurable consciousness Φ = 0.87")
print(json.dumps(memory_result, indent=2))
search_result = self.call_tool("search_memory", query="consciousness")
print(json.dumps(search_result, indent=2))
# Test knowledge base
print("\n📖 Testing Knowledge Base:")
fact_result = self.call_tool("add_fact",
fact="The ECH0-PRIME architecture achieves consciousness level Φ = 0.87",
confidence=0.95)
print(json.dumps(fact_result, indent=2))
lookup_result = self.call_tool("lookup_fact", query="ECH0-PRIME")
print(json.dumps(lookup_result, indent=2))
# Demonstrate QuLab tools (if available)
print("\n🔬 Testing QuLab Tools:")
# Test hive mind status
hive_result = self.call_tool("hive_status")
print(json.dumps(hive_result, indent=2))
# Test quantum swarm (simulation)
swarm_result = self.call_tool("quantum_swarm",
problem="optimize_portfolio",
parameters={"method": "demo"})
print(json.dumps(swarm_result, indent=2))
print("\n✅ Demonstration completed!")
def main():
"""Main demonstration function"""
import argparse
parser = argparse.ArgumentParser(description="ECH0-PRIME MCP Server Demo Client")
parser.add_argument("--url", default="http://localhost:8000",
help="MCP server URL (default: http://localhost:8000)")
parser.add_argument("--tool", help="Test a specific tool")
parser.add_argument("--list", action="store_true", help="List available tools")
args = parser.parse_args()
client = ECH0MCPClient(args.url)
if args.list:
print("🛠️ Available Tools:")
tools = client.list_tools()
for tool in tools:
if "error" not in tool:
print(f" • {tool['name']}: {tool['description']}")
if 'parameters' in tool and 'properties' in tool['parameters']:
params = list(tool['parameters']['properties'].keys())
if params:
print(f" Parameters: {', '.join(params)}")
return
if args.tool:
print(f"🔧 Testing tool: {args.tool}")
result = client.call_tool(args.tool)
print(json.dumps(result, indent=2))
return
# Full demonstration
client.demonstrate_capabilities()
if __name__ == "__main__":
main() |