#!/usr/bin/env python3 """Test MONIKA MCP server locally without Claude Desktop. This script demonstrates how to interact with MONIKA through MCP programmatically, useful for testing and debugging. """ import asyncio import json from pathlib import Path import pytest try: from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client except ImportError: print("MCP SDK not installed. Install with: pip install mcp") exit(1) async def _run_monika_mcp() -> None: """Exercise MONIKA MCP server via the MCP SDK.""" server_script = Path(__file__).parent.parent / "start_mcp_server.py" server_params = StdioServerParameters( command="python", args=[str(server_script)], env=None, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() print("=" * 60) print("MONIKA MCP Server Test") print("=" * 60) print("\n1. Listing available tools...") tools = await session.list_tools() print(f"Found {len(tools.tools)} tools:") for tool in tools.tools: print(f" - {tool.name}: {tool.description}") print("\n2. Getting training metrics...") result = await session.call_tool("get_training_metrics", {}) metrics = json.loads(result.content[0].text) print(f"Training step: {metrics['step']}") print(f"Vocab size: {metrics['vocab_size']}") print(f"Learning enabled: {metrics['learning_enabled']}") print("\n3. Getting yearning state...") result = await session.call_tool("yearning_state", {}) yearning = json.loads(result.content[0].text) print("Yearning dimensions:") for dim, values in yearning.items(): if isinstance(values, dict): y = values.get("yearning", 0) print(f" {dim}: yearning={y:.3f}") print("\n4. Getting controller dynamics...") result = await session.call_tool("controller_dynamics", {}) dynamics = json.loads(result.content[0].text) print(f"Last action: {dynamics.get('last_action', 'N/A')}") if "scores" in dynamics: print("Action scores:") for action, score in dynamics["scores"].items(): print(f" {action}: {score:.3f}") print("\n5. Getting memory snapshot...") result = await session.call_tool("memory_snapshot", {}) memory = json.loads(result.content[0].text) print(f"Facts: {len(memory.get('facts', []))}") print(f"Hypotheses: {len(memory.get('hypotheses', []))}") print(f"Todos: {len(memory.get('todos', []))}") print("\n6. Adding a memory fact...") result = await session.call_tool( "memory_apply", { "verb": { "op": "add_fact", "text": "Test fact from MCP client", "score": 1.0, } }, ) apply_result = json.loads(result.content[0].text) print(f"Applied: {apply_result['applied']}") print("\n7. Executing runtime step...") result = await session.call_tool( "runtime_step", {"text": "Hello MONIKA, this is a test message through MCP."}, ) step_result = json.loads(result.content[0].text) print(f"Step: {step_result['step']}") print(f"Verification passed: {step_result['verification_passed']}") print(f"Budget left: {step_result['budget_left']:.1f}") print("\n" + "=" * 60) print("Test complete! All tools working.") print("=" * 60) def test_monika_mcp(): """Synchronous pytest wrapper for the async MCP smoke test.""" pytest.importorskip("mcp") asyncio.run(_run_monika_mcp()) async def _run_interactive_loop() -> None: """Interactive REPL for manual MCP smoke testing.""" server_script = Path(__file__).parent.parent / "start_mcp_server.py" server_params = StdioServerParameters( command="python", args=[str(server_script)], env=None, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() print("=" * 60) print("MONIKA Interactive MCP Session") print("=" * 60) print("Commands:") print(" /yearning - Show yearning state") print(" /memory - Show memory snapshot") print(" /dynamics - Show controller dynamics") print(" /metrics - Show training metrics") print(" /step TEXT - Execute runtime step with TEXT") print(" /quit - Exit") print("=" * 60) while True: try: user_input = input("\n> ").strip() if not user_input: continue if user_input == "/quit": break if user_input == "/yearning": result = await session.call_tool("yearning_state", {}) yearning = json.loads(result.content[0].text) for dim, values in yearning.items(): if isinstance(values, dict): print(f"{dim}: yearning={values.get('yearning', 0):.3f}") elif user_input == "/memory": result = await session.call_tool("memory_snapshot", {}) memory = json.loads(result.content[0].text) print(json.dumps(memory, indent=2)) elif user_input == "/dynamics": result = await session.call_tool("controller_dynamics", {}) dynamics = json.loads(result.content[0].text) print(json.dumps(dynamics, indent=2, default=str)) elif user_input == "/metrics": result = await session.call_tool("get_training_metrics", {}) metrics = json.loads(result.content[0].text) print(json.dumps(metrics, indent=2)) elif user_input.startswith("/step "): text = user_input[6:] result = await session.call_tool("runtime_step", {"text": text}) step_result = json.loads(result.content[0].text) print(json.dumps(step_result, indent=2)) else: print(f"Unknown command: {user_input}") print("Use /quit to exit or try one of the commands above.") except KeyboardInterrupt: break except Exception as e: # pragma: no cover - manual REPL diagnostics print(f"Error: {e}") print("\nGoodbye!") @pytest.mark.skip(reason="Interactive MCP REPL is manual-only") def test_interactive_loop() -> None: """Placeholder so pytest reports the interactive REPL as skipped.""" pytest.importorskip("mcp") def main(): """CLI entry point.""" import sys if len(sys.argv) > 1 and sys.argv[1] == "--interactive": asyncio.run(_run_interactive_loop()) else: asyncio.run(_run_monika_mcp()) if __name__ == "__main__": main()