examples/demo_client.py

#!/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()