diff --git a/5-Applications/scripts/mcp_server.py b/5-Applications/scripts/mcp_server.py deleted file mode 100644 index d3037cfa..00000000 --- a/5-Applications/scripts/mcp_server.py +++ /dev/null @@ -1,163 +0,0 @@ -#!/usr/bin/env python3 -""" -Advanced MCP Server for Sovereign Research Stack - -Provides high-trust tools for: -- Swarm Intelligence (Active Questioning) -- Lean 4 Formal Verification (Consistentcy Checks) -- Manifold Topology Discovery (PIST Substrate) -- Academic Paper Context Extraction -""" - -import sys -import json -import asyncio -import subprocess -from pathlib import Path -from typing import Any, Dict, List, Optional - -# Add parent directory to path for imports -sys.path.insert(0, str(Path(__file__).parent)) -sys.path.insert(0, str(Path(__file__).parent / "scripts")) - -try: - from mcp.server import Server - from mcp.server.stdio import stdio_server - from mcp.types import Tool, TextContent -except ImportError: - print("MCP SDK not installed. Install with: pip install mcp") - sys.exit(1) - -# Import swarm components -try: - from scripts.enhanced_integrated_swarm import ( - EnhancedIntegratedSwarm, - create_demo_topology, - MathDatabase - ) -except ImportError: - print("Could not import swarm components") - sys.exit(1) - -# Global swarm instance -_swarm_instance: Optional[EnhancedIntegratedSwarm] = None - -def get_swarm() -> EnhancedIntegratedSwarm: - """Get or create swarm instance""" - global _swarm_instance - if _swarm_instance is None: - topology = create_demo_topology() - math_db = MathDatabase() - _swarm_instance = EnhancedIntegratedSwarm( - topology=topology, - math_db=math_db, - num_agents=50 - ) - return _swarm_instance - -# Create MCP server -server = Server("sovereign-research-stack") - -@server.list_tools() -async def list_tools() -> List[Tool]: - """List available advanced tools""" - return [ - Tool( - name="ask_swarm", - description="Consult the 50-agent swarm on complex reasoning tasks.", - inputSchema={ - "type": "object", - "properties": { - "question": {"type": "string"}, - "context": {"type": "string"}, - "domain": {"type": "string", "default": "theoretical_physics"} - }, - "required": ["question"] - } - ), - Tool( - name="query_manifold_topology", - description="Returns the current PIST-based virtual substrate state (Mass Field, Resonance).", - inputSchema={"type": "object", "properties": {}} - ), - Tool( - name="verify_lean_consistency", - description="Runs diagnostics on Lean modules to check for structural integrity.", - inputSchema={ - "type": "object", - "properties": { - "module": {"type": "string", "description": "Basename of the .lean file"} - }, - "required": ["module"] - } - ), - Tool( - name="get_academic_validation", - description="Retrieves arXiv-backed validation points for Semantic RG structures.", - inputSchema={"type": "object", "properties": {}} - ), - Tool( - name="teach_swarm_academic_papers", - description="Broadcasts academic paper content to the 50-agent swarm to update their research context.", - inputSchema={"type": "object", "properties": {}} - ) - ] - -@server.call_tool() -async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: - """Execute tool logic""" - swarm = get_swarm() - - if name == "ask_swarm": - response = swarm.research_api.ask_question( - question=arguments["question"], - context=arguments.get("context", ""), - domain=arguments.get("domain", "theoretical_physics") - ) - return [TextContent(type="text", text=f"Swarm Consensus:\n{response}")] - - elif name == "query_manifold_topology": - summary = swarm.optimizer.get_optimization_summary() - return [TextContent(type="text", text=json.dumps(summary, indent=2))] - - elif name == "verify_lean_consistency": - module = arguments["module"] - lean_path = Path("/home/allaun/Documents/Research Stack/0-Core-Formalism/lean/Semantics/Semantics") / module - if not lean_path.suffix == ".lean": - lean_path = lean_path.with_suffix(".lean") - - if lean_path.exists(): - # Perform pseudo-verification since we don't want to run full 'lake build' in a quick tool - with open(lean_path) as f: - content = f.read() - defs = content.count("def ") - theorems = content.count("theorem ") - sorries = content.count("sorry") - status = "Verified (Placeholder Sorries Exist)" if sorries > 0 else "Fully Proved" - return [TextContent(type="text", text=f"Module: {module}\nDefinitions: {defs}\nTheorems: {theorems}\nStatus: {status}")] - return [TextContent(type="text", text=f"Error: {module} not found.")] - - elif name == "get_academic_validation": - papers_file = Path("/home/allaun/Documents/Research Stack/shared-data/data/academic_papers_validation.json") - if papers_file.exists(): - with open(papers_file) as f: - data = json.load(f) - return [TextContent(type="text", text=json.dumps(data, indent=2))] - - elif name == "teach_swarm_academic_papers": - script_path = Path("/home/allaun/Documents/Research Stack/scripts/teach_swarm_academic_papers.py") - result = subprocess.run([sys.executable, str(script_path)], capture_output=True, text=True) - if result.returncode == 0: - return [TextContent(type="text", text=f"Swarm Teaching Successful:\n{result.stdout[-1000:]}")] - return [TextContent(type="text", text=f"Error teaching swarm:\n{result.stderr}")] - return [TextContent(type="text", text="Error: Validation data missing. Run teach_swarm script first.")] - - else: - raise ValueError(f"Unknown tool: {name}") - -async def main(): - async with stdio_server() as (read_stream, write_stream): - await server.run(read_stream, write_stream, server.create_initialization_options()) - -if __name__ == "__main__": - asyncio.run(main())