python_client.py

#!/usr/bin/env python3
"""
FORUS Cortex Agent MCP Server - Python Client Example
"""

import json
import requests
from typing import Dict, Any, Optional

class ForusMCPClient:
    """Python client for FORUS Cortex Agent MCP Server"""
    
    def __init__(self, server_url: str = "http://localhost:8083/mcp", api_key: Optional[str] = None):
        self.server_url = server_url
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"
        self._request_id = 0
    
    def _get_next_id(self) -> int:
        """Get next request ID"""
        self._request_id += 1
        return self._request_id
    
    def _make_request(self, method: str, params: Optional[Dict] = None) -> Dict[str, Any]:
        """Make MCP JSON-RPC request"""
        request = {
            "jsonrpc": "2.0",
            "id": self._get_next_id(),
            "method": method
        }
        
        if params:
            request["params"] = params
        
        try:
            response = requests.post(self.server_url, json=request, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            return {
                "jsonrpc": "2.0",
                "id": request["id"],
                "error": {
                    "code": -32603,
                    "message": f"Request failed: {str(e)}"
                }
            }
    
    def list_tools(self) -> Dict[str, Any]:
        """List all 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"""
        return self._make_request("tools/call", {
            "name": tool_name,
            "arguments": arguments
        })
    
    def search_memory(self, query: str, limit: int = 10, area: Optional[str] = None) -> Dict[str, Any]:
        """Search agent memory"""
        args = {"query": query, "limit": limit}
        if area:
            args["area"] = area
        return self.call_tool("memory_search", args)
    
    def query_msa(self, query: str) -> Dict[str, Any]:
        """Query Master Services Agreement documents"""
        return self.call_tool("query_msa", {"query": query})
    
    def execute_python(self, code: str, timeout: int = 30) -> Dict[str, Any]:
        """Execute Python code"""
        return self.call_tool("execute_python", {"code": code, "timeout": timeout})
    
    def execute_shell(self, command: str, timeout: int = 30) -> Dict[str, Any]:
        """Execute shell command"""
        return self.call_tool("execute_shell", {"command": command, "timeout": timeout})
    
    def get_system_status(self) -> Dict[str, Any]:
        """Get system status"""
        return self.call_tool("get_system_status", {})
    
    def list_mcp_tools(self) -> Dict[str, Any]:
        """List MCP tools with details"""
        return self.call_tool("list_mcp_tools", {})

def print_response(response: Dict[str, Any], title: str = "Response"):
    """Pretty print MCP response"""
    print(f"\n{'='*50}")
    print(f"{title}")
    print('='*50)
    
    if "error" in response:
        print(f"❌ ERROR: {response['error']['message']}")
        return
    
    if "result" in response:
        result = response["result"]
        if "content" in result and isinstance(result["content"], list):
            # Tool call response
            for content in result["content"]:
                if content.get("type") == "text":
                    try:
                        data = json.loads(content["text"])
                        print(json.dumps(data, indent=2))
                    except json.JSONDecodeError:
                        print(content["text"])
        else:
            # Direct result
            print(json.dumps(result, indent=2))

def main():
    """Example usage of the FORUS MCP Client"""
    
    # Initialize client (change URL to match your server)
    client = ForusMCPClient("http://93.127.162.206:8083/mcp")
    
    # List available tools
    print("🔧 Listing available tools...")
    tools_response = client.list_tools()
    print_response(tools_response, "Available Tools")
    
    # Search memory
    print("\n🧠 Searching agent memory...")
    memory_response = client.search_memory("FORUS business model", limit=3)
    print_response(memory_response, "Memory Search Results")
    
    # Query MSA documents
    print("\n📋 Querying MSA documents...")
    msa_response = client.query_msa("system operator obligations")
    print_response(msa_response, "MSA Query Results")
    
    # Execute Python code
    print("\n🐍 Executing Python code...")
    python_response = client.execute_python("""
import datetime
import platform

print(f"Hello from FORUS MCP Server!")
print(f"Current time: {datetime.datetime.now()}")
print(f"Platform: {platform.system()} {platform.release()}")
print(f"Python version: {platform.python_version()}")

# Simple calculation
result = 42 * 2
print(f"The answer to everything times 2 is: {result}")
""")
    print_response(python_response, "Python Execution Results")
    
    # Execute shell command
    print("\n💻 Executing shell command...")
    shell_response = client.execute_shell("echo 'Hello from shell!' && date && pwd")
    print_response(shell_response, "Shell Command Results")
    
    # Get system status
    print("\n📊 Getting system status...")
    status_response = client.get_system_status()
    print_response(status_response, "System Status")
    
    # List MCP tools with details
    print("\n🛠️ Listing MCP tools with details...")
    mcp_tools_response = client.list_mcp_tools()
    print_response(mcp_tools_response, "MCP Tools Details")
    
    print("\n✅ All examples completed successfully!")
    print("\n💡 You can now use this client in your own applications:")
    print("    client = ForusMCPClient('http://your-server:8083/mcp')")
    print("    result = client.search_memory('your query')")

if __name__ == "__main__":
    main()