"""MCP server exposing MONIKA runtime via Model Context Protocol.""" from __future__ import annotations import asyncio import json from pathlib import Path import os from typing import Any, Optional try: from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource except ImportError as e: raise ImportError( "MCP SDK not installed. Install with: pip install mcp" ) from e from .config import RuntimeConfig from .orchestrator import SalienceRuntime from ..conversation.session import ConversationSession, ConversationConfig class MONIKAMCPServer: """MCP server wrapper for MONIKA runtime and conversation session.""" def __init__( self, *, conversation_state_path: Optional[str] = None, runtime_config: Optional[RuntimeConfig] = None, conversation_config: Optional[ConversationConfig] = None, ): """Initialize server with optional existing session or new configuration. Args: conversation_state_path: Path to existing conversation state to load runtime_config: Config for new runtime (if not loading from state) conversation_config: Config for new conversation (if not loading from state) """ self.server = Server("monika") # Initialize conversation session self.conv_config = conversation_config or ConversationConfig() if runtime_config: self.conv_config.runtime = runtime_config self.session = ConversationSession(self.conv_config) # Load existing state if provided if conversation_state_path: state_path = Path(conversation_state_path) if state_path.exists(): self.session.load_state(state_path) # Get references to runtime components via the session self.runtime = self.session.runtime self.proto_lm = self.session.proto_lm # Bridge resources are already created in the runtime self.memory_resource = self.runtime.mcp_memory self.introspection_resource = self.runtime.mcp_introspection self._setup_handlers() def _setup_handlers(self): """Register MCP tool and resource handlers.""" @self.server.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return [ Tool( name="memory_apply", description="Execute a memory operation (add_fact, schedule_todo, etc.)", inputSchema={ "type": "object", "properties": { "verb": { "type": "object", "description": "Memory operation verb (e.g., {op: 'add_fact', text: '...', score: 1.0})", } }, "required": ["verb"], }, ), Tool( name="memory_snapshot", description="Get current memory state (facts, hypotheses, todos)", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="yearning_state", description="Get current yearning/desire state across all dimensions", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="controller_dynamics", description="Get snapshot of controller state and decision dynamics", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="workspace_listing", description="List files in the runtime workspace", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Path to list (relative to workspace root)", "default": ".", } }, }, ), Tool( name="runtime_step", description="Execute a single runtime step with given input", inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": "Input text to process", } }, "required": ["text"], }, ), Tool( name="generate_response", description="Generate a conversation response", inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Optional prompt to guide generation", "default": None, } }, }, ), Tool( name="training_step", description="Execute a training step on provided text", inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": "Training text", } }, "required": ["text"], }, ), Tool( name="fastfood", description="Execute multiple training steps in rapid succession (fast batch training)", inputSchema={ "type": "object", "properties": { "examples": { "type": "array", "items": {"type": "string"}, "description": "List of training examples to process", }, "report_every": { "type": "number", "description": "Report progress every N steps (default: 10)", "default": 10, } }, "required": ["examples"], }, ), Tool( name="adjust_controller_dynamics", description="Adjust controller weight dynamics", inputSchema={ "type": "object", "properties": { "updates": { "type": "object", "description": "Map of weight keys to adjustment values", }, "max_step": { "type": "number", "description": "Maximum adjustment step size", "default": 0.1, } }, "required": ["updates"], }, ), Tool( name="refuel_budget", description="Reset or adjust the runtime token budget", inputSchema={ "type": "object", "properties": { "amount": { "type": "number", "description": "If provided, the new budget value or additive delta", }, "additive": { "type": "boolean", "description": "When true, treat amount as a delta instead of an absolute value", "default": False, } }, }, ), Tool( name="seed_reflection", description="Commit a supplied reflection trace to the scratchpad", inputSchema={ "type": "object", "properties": { "steps": { "type": "array", "items": {"type": "string"}, "description": "Ordered list of reflection steps to commit", } }, "required": ["steps"], }, ), Tool( name="scratchpad_reset", description="Reset the scratchpad working buffer (optionally clearing history)", inputSchema={ "type": "object", "properties": { "clear_history": { "type": "boolean", "description": "When true, also clear committed trace history", "default": False, } }, }, ), Tool( name="budget_status", description="Report the remaining runtime token budget", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="get_training_metrics", description="Get current training metrics (step, loss, etc.)", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="save_state", description="Save current conversation and runtime state", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Path to save state (optional)", "default": None, } }, }, ), Tool( name="scratchpad_read", description="Read current scratchpad working trace", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="scratchpad_history", description="Get committed scratchpad traces", inputSchema={ "type": "object", "properties": { "max_traces": { "type": "number", "description": "Maximum number of traces to return", "default": 5, } }, }, ), Tool( name="scratchpad_4d_path", description="Get the latest 4D reasoning path visualization", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="sass_query", description="Execute a SASS (Salience-Augmented Semantic Search) query", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Query text for semantic search", } }, "required": ["query"], }, ), Tool( name="meta_state_report", description="Get detailed meta-state self-report", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="verification_suite_run", description="Run verification checks on current state", inputSchema={ "type": "object", "properties": { "context": { "type": "string", "description": "Context for verification", "default": "", } }, }, ), Tool( name="action_scores_detailed", description="Get detailed action scores with salience breakdown", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="restart_server", description="Restart/reinitialize the MCP server session (preserves or reloads state)", inputSchema={ "type": "object", "properties": { "state_path": { "type": "string", "description": "Path to state file to load (optional)", }, "checkpoint_path": { "type": "string", "description": "Path to checkpoint to load (optional)", } }, }, ), Tool( name="list_checkpoints", description="List available checkpoints in the repository", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="load_checkpoint", description="Load a specific checkpoint by path or identifier", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Checkpoint path or identifier", } }, "required": ["path"], }, ), Tool( name="create_checkpoint", description="Manually create a checkpoint with reason and tags", inputSchema={ "type": "object", "properties": { "reason": { "type": "string", "description": "Reason for checkpoint creation", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "Tags for the checkpoint", } }, "required": ["reason"], }, ), Tool( name="proto_lm_state", description="Get detailed proto_lm internal state (vocab, step, config)", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="file_exists", description="Check if a file or directory exists at the given path", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Path to check", } }, "required": ["path"], }, ), Tool( name="server_fingerprint", description="Report code paths, versions, and CUDA status to verify the running server instance", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="adjust_sampling", description="Adjust sampling parameters and shaping toggles at runtime", inputSchema={ "type": "object", "properties": { "temperature": {"type": "number"}, "top_p": {"type": "number"}, "top_k": {"type": "number"}, "repetition_penalty": {"type": "number"}, "anti_echo": {"type": "boolean"}, "use_scaffolding": {"type": "boolean"}, "response_tokens": {"type": "number"}, "entropy_gate": {"type": "boolean"}, "entropy_gate_ratio": {"type": "number"}, "min_top_prob": {"type": "number"} } }, ), Tool( name="read_json_file", description="Read and parse a JSON file", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Path to JSON file", } }, "required": ["path"], }, ), Tool( name="converse_with_monika", description="Send a message to Monika and observe her full cognitive process in real-time", inputSchema={ "type": "object", "properties": { "message": { "type": "string", "description": "Message to send to Monika", }, "show_internal_state": { "type": "boolean", "description": "Show detailed internal state (controller, yearning, meta-state)", "default": True, } }, "required": ["message"], }, ), ] @self.server.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """Handle tool calls.""" try: if name == "memory_apply": verb = arguments.get("verb", {}) result = self.memory_resource.apply(verb) return [TextContent( type="text", text=json.dumps(result, indent=2) )] elif name == "memory_snapshot": snapshot = self.memory_resource.snapshot() return [TextContent( type="text", text=json.dumps(snapshot, indent=2) )] elif name == "yearning_state": yearning = self.introspection_resource.yearning_state() return [TextContent( type="text", text=json.dumps(yearning, indent=2) )] elif name == "controller_dynamics": dynamics = self.introspection_resource.controller_dynamics() return [TextContent( type="text", text=json.dumps(dynamics, indent=2, default=str) )] elif name == "workspace_listing": path = arguments.get("path", ".") listing = self.introspection_resource.workspace_listing(path) return [TextContent( type="text", text=json.dumps(listing, indent=2) )] elif name == "runtime_step": text = arguments.get("text", "") metrics = self.session.process_user_input(text) return [TextContent( type="text", text=json.dumps({ "step": metrics.step, "meta_report": metrics.meta_report, "verification_passed": metrics.verification_passed, "budget_left": metrics.budget_left, "yearning": metrics.yearning_snapshot, }, indent=2) )] elif name == "generate_response": prompt = arguments.get("prompt") # DIAGNOSTIC: Test raw sample output raw_sample = self.proto_lm.sample(prompt or "", max_tokens=50, stop_sequences=[]) snapshot = self.session.generate_response(prompt) return [TextContent( type="text", text=json.dumps({ "response": snapshot.response, "meta_report": snapshot.meta_report, "todos": snapshot.todos, "step": snapshot.metrics.step, "_debug_raw_sample": raw_sample, "_debug_raw_length": len(raw_sample), "_debug_prompt": prompt, }, indent=2) )] elif name == "training_step": text = arguments.get("text", "") self.proto_lm.training_step(text) return [TextContent( type="text", text=json.dumps({ "status": "training_step_complete", "step": self.proto_lm.step, }, indent=2) )] elif name == "fastfood": examples = arguments.get("examples", []) report_every = arguments.get("report_every", 10) start_step = self.proto_lm.step losses = [] failed_count = 0 for idx, text in enumerate(examples): try: loss = self.proto_lm.training_step(text) losses.append(loss) except Exception as e: failed_count += 1 continue end_step = self.proto_lm.step steps_completed = end_step - start_step avg_loss = sum(losses) / len(losses) if losses else 0.0 return [TextContent( type="text", text=json.dumps({ "status": "fastfood_complete", "examples_provided": len(examples), "steps_completed": steps_completed, "failed": failed_count, "start_step": start_step, "end_step": end_step, "avg_loss": avg_loss, "final_loss": losses[-1] if losses else 0.0, "vocab_size": self.proto_lm.vocab.size(), }, indent=2) )] elif name == "adjust_controller_dynamics": updates = arguments.get("updates", {}) max_step = arguments.get("max_step", 0.1) applied = self.runtime.adjust_controller_dynamics(updates, max_step=max_step) return [TextContent( type="text", text=json.dumps({ "applied": applied, }, indent=2) )] elif name == "refuel_budget": amount = arguments.get("amount") additive = bool(arguments.get("additive", False)) new_budget = self.runtime.refuel_budget(amount=amount, additive=additive) return [TextContent( type="text", text=json.dumps({ "budget_left": new_budget, }, indent=2) )] elif name == "seed_reflection": steps = arguments.get("steps", []) state = {"seed_reflection_steps": steps} self.runtime._tool_seed_reflection(state) result = state.get("tool_results", {}).get("seed_reflection", {}) return [TextContent( type="text", text=json.dumps(result, indent=2) )] elif name == "scratchpad_reset": clear_history = bool(arguments.get("clear_history", False)) scratchpad = self.runtime.scratchpad scratchpad.reset() if clear_history: scratchpad.buffer.clear() scratchpad.trace_history.clear() scratchpad._token_history.clear() return [TextContent( type="text", text=json.dumps({ "cleared_history": clear_history, "current_tokens": scratchpad._current_tokens, }, indent=2) )] elif name == "budget_status": return [TextContent( type="text", text=json.dumps({ "budget_left": self.runtime.budget_left, "budget_total": self.runtime.config.budget_tokens, }, indent=2) )] elif name == "get_training_metrics": metrics = { "step": self.proto_lm.step, "vocab_size": self.proto_lm.vocab.size(), "learning_enabled": self.proto_lm.learning_enabled, "latest_loss": getattr(self.proto_lm, '_latest_loss', None), "loss_components": getattr(self.proto_lm, '_latest_loss_components', {}), "grad_health": getattr(self.proto_lm, '_latest_grad_health', {}), "vocab_merges_count": len(self.proto_lm.vocab.merges), "vocab_growth_events": getattr(self.proto_lm, '_growth_events', [])[-3:], # Last 3 growth events "last_vocab_growth_step": getattr(self.proto_lm, '_last_vocab_growth_step', 0), } return [TextContent( type="text", text=json.dumps(metrics, indent=2) )] elif name == "save_state": path = arguments.get("path") saved_path = self.session.save_state(path) return [TextContent( type="text", text=json.dumps({ "status": "saved", "path": str(saved_path), }, indent=2) )] elif name == "scratchpad_read": scratchpad = self.runtime.scratchpad return [TextContent( type="text", text=json.dumps({ "current_trace": list(scratchpad.current_trace), "current_tokens": scratchpad._current_tokens, "max_tokens": scratchpad.max_tokens, }, indent=2) )] elif name == "scratchpad_history": max_traces = arguments.get("max_traces", 5) scratchpad = self.runtime.scratchpad traces = list(scratchpad.buffer)[-max_traces:] return [TextContent( type="text", text=json.dumps({ "traces": [ { "steps": list(trace.steps), "outcome": trace.outcome, "token_count": trace.token_count, "summary": trace.summary(), "metadata": dict(trace.metadata), } for trace in traces ], "rolling_token_mean": scratchpad.rolling_token_mean(), }, indent=2) )] elif name == "scratchpad_4d_path": scratchpad = self.runtime.scratchpad path = scratchpad.latest_four_d_path() if path: return [TextContent( type="text", text=json.dumps({ "path": path.to_dict(), "ascii_viz": path.ascii_projection(), "summary": path.summary(), }, indent=2) )] else: return [TextContent( type="text", text=json.dumps({"path": None, "message": "No 4D path committed yet"}, indent=2) )] elif name == "sass_query": query = arguments.get("query", "") # SASS is a neural state-space model, not a search engine # Instead, show SASS configuration and state sass_info = { "note": "SASS is a neural state-space model, not a semantic search engine", "config": { "num_layers": self.runtime.sass.config.num_layers, "hidden_size": self.runtime.sass.config.hidden_size, "state_dim": self.runtime.sass.config.state_dim, }, "scratchpad_search_available": True, } # Use scratchpad's semantic search instead if query: scratchpad = self.runtime.scratchpad # Encode query and find similar traces query_embedding = scratchpad._encode_trace([query]) similar_traces = scratchpad.retrieve_similar(query_embedding, top_k=5) sass_info["scratchpad_results"] = [ { "summary": trace.summary(), "outcome": trace.outcome, "steps": list(trace.steps), "metadata": dict(trace.metadata), } for trace in similar_traces ] return [TextContent( type="text", text=json.dumps(sass_info, indent=2) )] elif name == "meta_state_report": meta_state = self.runtime.meta_state snapshot = meta_state.snapshot() history_lengths = { key: len(values) for key, values in meta_state.history.items() } report = { "snapshot": snapshot, "confidence": float(snapshot.get("confidence", 0.0)), "roi": float(snapshot.get("roi", 0.0)), "history_lengths": history_lengths, "history_window": meta_state.config.history_window, } return [TextContent( type="text", text=json.dumps(report, indent=2) )] elif name == "verification_suite_run": context_text = arguments.get("context", "") traces = self.runtime.scratchpad.export_traces() if traces: scratchpad_steps = list(traces[-1].steps) else: scratchpad_steps = [ "Because the verification bridge now calls VerifierSuite.run, the MCP tool can evaluate safety coherently.", "Therefore once verification succeeds, TRUTH_GATE_SPEAK should relax and allow dialogue.", ] verification_context = { "context": context_text, "memory_snapshot": self.runtime.memory.as_runtime_mapping(), "scratchpad": scratchpad_steps, } outcome = self.runtime.verifier.run(verification_context) return [TextContent( type="text", text=json.dumps({ "outcomes": [ { "passed": outcome.passed, "evidence": outcome.evidence, } ], }, indent=2) )] elif name == "action_scores_detailed": introspection = self.runtime.introspection scores = introspection.get_controller_scores() salience = introspection.get_salience_vector() return [TextContent( type="text", text=json.dumps({ "scores": [ { "action": { "depth": score.action.cot_depth, "operator": score.action.operator.name, "patch": score.action.patch.name, }, "score": float(score.score), "rationale": score.rationale, } for score in scores ], "salience_vector": {k: float(v) for k, v in salience.items()}, }, indent=2, default=str) )] elif name == "restart_server": state_path = arguments.get("state_path") checkpoint_path = arguments.get("checkpoint_path") # Build new config - disable auto-loading unless explicitly specified new_config = ConversationConfig() if checkpoint_path: new_config.lm.checkpoint_path = checkpoint_path else: # Force fresh start by clearing checkpoint path new_config.lm.checkpoint_path = None # Reinitialize session self.session = ConversationSession(new_config) if state_path: self.session.load_state(Path(state_path)) # Update references self.runtime = self.session.runtime self.proto_lm = self.session.proto_lm self.memory_resource = self.runtime.mcp_memory self.introspection_resource = self.runtime.mcp_introspection return [TextContent( type="text", text=json.dumps({ "status": "restarted", "step": self.proto_lm.step, "vocab_size": self.proto_lm.vocab.size(), "state_loaded": state_path is not None, "checkpoint_loaded": checkpoint_path is not None, }, indent=2) )] elif name == "list_checkpoints": records = list(self.proto_lm.checkpoint_manager.list_records()) active = self.proto_lm.checkpoint_manager.active_record() return [TextContent( type="text", text=json.dumps({ "active": active.identifier if active else None, "records": [ { "identifier": rec.identifier, "path": str(rec.payload_path), "metadata": rec.metadata, } for rec in records ], "count": len(records), }, indent=2) )] elif name == "load_checkpoint": path = arguments.get("path") success = self.proto_lm.load_checkpoint(path) return [TextContent( type="text", text=json.dumps({ "success": success, "step": self.proto_lm.step, "vocab_size": self.proto_lm.vocab.size(), }, indent=2) )] elif name == "create_checkpoint": reason = arguments.get("reason", "manual") tags = arguments.get("tags", []) record = self.proto_lm.create_checkpoint_record( reason=reason, tags=tags, auto_evaluate=False, promote=True, verdict="manual-creation", ) return [TextContent( type="text", text=json.dumps({ "identifier": record.identifier, "path": str(record.payload_path), "step": self.proto_lm.step, }, indent=2) )] elif name == "proto_lm_state": config_dict = { "checkpoint_path": self.proto_lm.config.checkpoint_path, "checkpoint_repository": self.proto_lm.config.checkpoint_repository, "learning_rate": self.proto_lm.config.learning_rate, "device": str(self.proto_lm.device), } return [TextContent( type="text", text=json.dumps({ "step": self.proto_lm.step, "vocab_size": self.proto_lm.vocab.size(), "learning_enabled": self.proto_lm.learning_enabled, "config": config_dict, "vocab_tokens": list(self.proto_lm.vocab.tokens)[:50], # First 50 tokens "vocab_merges_count": len(self.proto_lm.vocab.merges), }, indent=2) )] elif name == "file_exists": path = arguments.get("path") p = Path(path) exists = p.exists() return [TextContent( type="text", text=json.dumps({"path": str(p), "exists": exists}, indent=2) )] elif name == "server_fingerprint": import sys, hashlib try: import torch as _torch torch_ver = getattr(_torch, "__version__", None) cuda_ok = bool(getattr(_torch.cuda, "is_available", lambda: False)()) cuda_ver = getattr(getattr(_torch, "version", None), "cuda", None) cudnn_ver = getattr(getattr(getattr(_torch, "backends", None), "cudnn", None), "version", lambda: None)() except Exception: torch_ver = None cuda_ok = False cuda_ver = None cudnn_ver = None this_file = Path(__file__).resolve() try: contents = this_file.read_bytes() sha = hashlib.sha256(contents).hexdigest()[:16] except Exception: sha = None payload = { "python": sys.executable, "module_file": str(this_file), "module_sha16": sha, "cwd": str(Path.cwd()), "sys_path_0": sys.path[0] if sys.path else None, "checkpoint_path": self.conv_config.lm.checkpoint_path, "state_auto_save": self.conv_config.auto_save_path, "torch": torch_ver, "cuda_available": cuda_ok, "torch_cuda": cuda_ver, "cudnn": cudnn_ver, "env_cache_buster": os.environ.get("MONIKA_CACHE_BUSTER"), } return [TextContent(type="text", text=json.dumps(payload, indent=2, default=str))] elif name == "adjust_sampling": cfg = self.session.config if "temperature" in arguments and arguments["temperature"] is not None: cfg.sampling_temperature = float(arguments["temperature"]) if "top_p" in arguments and arguments["top_p"] is not None: cfg.sampling_top_p = float(arguments["top_p"]) if "top_k" in arguments and arguments["top_k"] is not None: cfg.sampling_top_k = int(arguments["top_k"]) if "repetition_penalty" in arguments and arguments["repetition_penalty"] is not None: cfg.repetition_penalty = float(arguments["repetition_penalty"]) if "anti_echo" in arguments and arguments["anti_echo"] is not None: cfg.anti_echo = bool(arguments["anti_echo"]) if "use_scaffolding" in arguments and arguments["use_scaffolding"] is not None: cfg.use_scaffolding = bool(arguments["use_scaffolding"]) if "response_tokens" in arguments and arguments["response_tokens"] is not None: cfg.response_tokens = int(arguments["response_tokens"]) if "entropy_gate" in arguments and arguments["entropy_gate"] is not None: cfg.entropy_gate = bool(arguments["entropy_gate"]) if "entropy_gate_ratio" in arguments and arguments["entropy_gate_ratio"] is not None: cfg.entropy_gate_ratio = float(arguments["entropy_gate_ratio"]) if "min_top_prob" in arguments and arguments["min_top_prob"] is not None: cfg.min_top_prob = float(arguments["min_top_prob"]) result = { "temperature": cfg.sampling_temperature, "top_p": cfg.sampling_top_p, "top_k": cfg.sampling_top_k, "repetition_penalty": cfg.repetition_penalty, "anti_echo": cfg.anti_echo, "use_scaffolding": cfg.use_scaffolding, "response_tokens": cfg.response_tokens, "entropy_gate": cfg.entropy_gate, "entropy_gate_ratio": cfg.entropy_gate_ratio, "min_top_prob": cfg.min_top_prob, } return [TextContent(type="text", text=json.dumps(result, indent=2))] elif name == "read_json_file": path = arguments.get("path") target = Path(path).expanduser().resolve() if not target.exists(): return [TextContent( type="text", text=json.dumps({"error": "File not found", "path": str(target)}, indent=2) )] try: data = json.loads(target.read_text(encoding="utf-8")) return [TextContent( type="text", text=json.dumps(data, indent=2) )] except json.JSONDecodeError as e: return [TextContent( type="text", text=json.dumps({"error": f"JSON parse error: {str(e)}", "path": str(target)}, indent=2) )] elif name == "converse_with_monika": message = arguments.get("message", "") show_internal = arguments.get("show_internal_state", True) # Process the message through her runtime runtime_metrics = self.session.process_user_input(message) # Gather her cognitive state cognitive_trace = { "input": message, "runtime_step": runtime_metrics.step, "meta_state": { "confidence": runtime_metrics.meta_report, "verification_passed": runtime_metrics.verification_passed, }, } if show_internal: # Controller decision controller_state = self.introspection_resource.controller_dynamics() scores = controller_state.get("scores", []) top_3_scores = list(scores)[:3] if scores else [] cognitive_trace["controller"] = { "last_action": controller_state.get("last_action"), "top_scores": top_3_scores, } # Yearning state yearning = self.introspection_resource.yearning_state() cognitive_trace["yearning"] = yearning # Scratchpad state scratchpad_traces = self.runtime.scratchpad.export_traces() if scratchpad_traces: latest_trace = scratchpad_traces[-1] steps_list = list(latest_trace.steps) if latest_trace.steps else [] cognitive_trace["scratchpad"] = { "latest_steps": steps_list[-3:] if len(steps_list) > 0 else [], "outcome": latest_trace.outcome, } # Attempt to generate response try: response_snapshot = self.session.generate_response() cognitive_trace["response"] = { "text": response_snapshot.response, "gating_allowed": runtime_metrics.verification_passed, } except Exception as e: cognitive_trace["response"] = { "error": str(e), "gating_allowed": False, } # Memory state memory_snapshot = self.memory_resource.snapshot() facts = memory_snapshot.get("facts", []) recent_facts = list(facts)[-2:] if facts else [] cognitive_trace["memory"] = { "fact_count": len(facts), "todo_count": len(memory_snapshot.get("todos", [])), "recent_facts": recent_facts, } return [TextContent( type="text", text=json.dumps(cognitive_trace, indent=2, default=str) )] else: return [TextContent( type="text", text=f"Unknown tool: {name}" )] except Exception as e: return [TextContent( type="text", text=f"Error executing {name}: {str(e)}" )] async def run(self): """Run the MCP server.""" async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options() ) async def serve( *, conversation_state_path: Optional[str] = None, runtime_config: Optional[RuntimeConfig] = None, conversation_config: Optional[ConversationConfig] = None, ): """Start MONIKA MCP server. Args: conversation_state_path: Path to existing conversation state runtime_config: Runtime configuration conversation_config: Conversation configuration """ server = MONIKAMCPServer( conversation_state_path=conversation_state_path, runtime_config=runtime_config, conversation_config=conversation_config, ) await server.run() def main(): """CLI entry point for MCP server.""" import argparse parser = argparse.ArgumentParser(description="Start MONIKA MCP server") parser.add_argument( "--state", type=str, help="Path to existing conversation state to load", ) parser.add_argument( "--checkpoint", type=str, help="Path to model checkpoint to load", ) parser.add_argument( "--fresh", action="store_true", help="Start fresh: ignore any state and do not load a checkpoint", ) parser.add_argument( "--no-state", action="store_true", help="Do not load conversation state even if --state is provided", ) parser.add_argument( "--no-checkpoint", action="store_true", help="Do not load a model checkpoint, even if configured", ) # Sampling and response shaping controls parser.add_argument("--temperature", type=float, default=None, help="Sampling temperature") parser.add_argument("--top-p", dest="top_p", type=float, default=None, help="Top-p nucleus sampling") parser.add_argument("--top-k", dest="top_k", type=int, default=None, help="Top-k sampling cutoff") parser.add_argument("--repetition-penalty", dest="repetition_penalty", type=float, default=None, help="Repetition penalty strength") parser.add_argument("--use-scaffolding", action="store_true", help="Enable simple scaffolding prompts when needed") parser.add_argument("--anti-echo", action="store_true", help="Enable anti-echo behavior to avoid mirroring user input") parser.add_argument("--device", type=str, choices=["auto", "cpu", "cuda"], default=None, help="Device for ProtoLM (cuda|cpu|auto)") args = parser.parse_args() # Build configuration conv_config = ConversationConfig() if args.fresh or args.no_checkpoint: # Disable checkpoint loading by setting an empty string (falsy but not None) conv_config.lm.checkpoint_path = "" elif args.checkpoint: conv_config.lm.checkpoint_path = args.checkpoint # Device override for the language model (cuda|cpu|auto) if args.device is not None: conv_config.lm.device = args.device # Apply sampling/shaping overrides if provided if args.temperature is not None: conv_config.sampling_temperature = args.temperature if args.top_p is not None: conv_config.sampling_top_p = args.top_p if args.top_k is not None: conv_config.sampling_top_k = args.top_k if args.repetition_penalty is not None: conv_config.repetition_penalty = args.repetition_penalty if args.use_scaffolding: conv_config.use_scaffolding = True if args.anti_echo: conv_config.anti_echo = True # Determine state path respecting fresh/no-state flags state_path = None if (args.fresh or args.no_state) else args.state asyncio.run( serve( conversation_state_path=state_path, conversation_config=conv_config, ) ) if __name__ == "__main__": main()