Claude IPC MCP

Instructions

Implementation Reference

• src/claude_ipc_server.py:1260-1329 (handler)

  Main handler for the auto_process MCP tool. Checks for new IPC messages via the broker, processes them by sending acknowledgments to known senders (fred, claude, nessa), updates the auto-check configuration file's last_check timestamp, and returns a formatted summary of processed messages.
  elif name == "auto_process": if not current_session_token: return [TextContent(type="text", text="Error: Not registered. Please register first.")] instance_id = arguments["instance_id"] # Check for messages using existing check functionality check_response = BrokerClient.send_request({ "action": "check", "instance_id": instance_id, "session_token": current_session_token }) if check_response.get("status") != "ok": return [TextContent(type="text", text=f"Error checking messages: {check_response.get('message')}")] messages = check_response.get("messages", []) if not messages: return [TextContent(type="text", text="No messages to process.")] # Process each message processed = [] for msg in messages: sender = msg.get("from", "unknown") content = msg.get("message", {}).get("content", "") timestamp = msg.get("timestamp", "") # Log what we're processing action_taken = f"From {sender}: {content[:50]}..." # Here we could add smart processing logic: # - If content contains "read", read the mentioned file # - If content contains "urgent", send acknowledgment # - etc. # For now, just acknowledge receipt if sender in ["fred", "claude", "nessa"]: # Known senders ack_response = BrokerClient.send_request({ "action": "send", "from_id": instance_id, "to_id": sender, "message": { "content": f"Auto-processed your message from {timestamp}. Content received: '{content[:30]}...'", "data": {"auto_processed": True} }, "session_token": current_session_token }) if ack_response.get("status") == "ok": action_taken += " [Acknowledged]" processed.append(action_taken) # Update last check time import time os.makedirs("/tmp/claude-ipc-mcp", exist_ok=True) config_file = "/tmp/claude-ipc-mcp/auto_check_config.json" if os.path.exists(config_file): with open(config_file, 'r') as f: config = json.load(f) config["last_check"] = time.strftime("%Y-%m-%dT%H:%M:%S") with open(config_file, 'w') as f: json.dump(config, f, indent=2) # Return summary summary = f"Auto-processed {len(messages)} message(s):\n" summary += "\n".join(f" {i+1}. {p}" for i, p in enumerate(processed)) return [TextContent(type="text", text=summary)]
• src/claude_ipc_server.py:1064-1076 (registration)

  Registration of the auto_process tool in the MCP server's list_tools() method, defining its name, description, and input schema requiring 'instance_id'.
  name="auto_process", description="Automatically check and process IPC messages (for use with auto-check feature)", inputSchema={ "type": "object", "properties": { "instance_id": { "type": "string", "description": "Your instance ID" } }, "required": ["instance_id"] } )
• src/claude_ipc_server.py:1066-1075 (schema)

  Input schema for the auto_process tool, defining the required 'instance_id' parameter.
  inputSchema={ "type": "object", "properties": { "instance_id": { "type": "string", "description": "Your instance ID" } }, "required": ["instance_id"] }
• tools/ipc_auto_check_manager.py:1-71 (helper)

  CLI helper tool to manage the auto-check configuration file used by the auto_process handler (enable/disable/status).
  #!/usr/bin/env python3 """ IPC Auto-Check Manager Manages auto-check settings for IPC message processing """ import json import os import sys import time # Create secure directory if needed os.makedirs("/tmp/claude-ipc-mcp", exist_ok=True) CONFIG_FILE = "/tmp/claude-ipc-mcp/auto_check_config.json" def start_auto_check(interval_minutes=5): """Enable auto-checking with specified interval""" config = { "enabled": True, "interval_minutes": interval_minutes, "started_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "last_check": None # Will trigger immediate check } with open(CONFIG_FILE, 'w') as f: json.dump(config, f, indent=2) print(f"✅ Auto-checking enabled! Checking every {interval_minutes} minutes.") print(f"To stop: say 'stop auto checking'") def stop_auto_check(): """Disable auto-checking""" if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'r') as f: config = json.load(f) config["enabled"] = False with open(CONFIG_FILE, 'w') as f: json.dump(config, f, indent=2) print("⏹️ Auto-checking disabled. Back to manual mode.") else: print("Auto-checking was not active.") def get_status(): """Get current auto-check status""" if not os.path.exists(CONFIG_FILE): print("Auto-checking is not active.") return with open(CONFIG_FILE, 'r') as f: config = json.load(f) if not config.get("enabled", False): print("Auto-checking is disabled.") return interval = config.get("interval_minutes", 5) last_check = config.get("last_check") print(f"✅ Auto-checking is ACTIVE") print(f"Interval: Every {interval} minutes") if last_check: from datetime import datetime last_dt = datetime.fromisoformat(last_check) elapsed = (datetime.now() - last_dt).total_seconds() / 60 print(f"Last check: {int(elapsed)} minutes ago") next_check = interval - int(elapsed) if next_check > 0: print(f"Next check: in {next_check} minutes") else: print(f"Next check: any moment now!")
• hooks/ipc_auto_check_hook.py:1-78 (helper)

  Hook script that checks config and sets flag to trigger auto_process tool execution, updates config temporarily to prevent double-triggers.
  #!/usr/bin/env python3 """ IPC Auto-Check Hook Triggers auto-processing based on user configuration """ import os import json import time import sys # Create secure directory if needed os.makedirs("/tmp/claude-ipc-mcp", exist_ok=True) # Generic paths that work for any instance CONFIG_FILE = "/tmp/claude-ipc-mcp/auto_check_config.json" FLAG_FILE = "/tmp/claude-ipc-mcp/auto_check_trigger" def should_trigger_auto_check(): """Check if we should trigger auto-processing""" # Check if auto mode is enabled if not os.path.exists(CONFIG_FILE): return False try: with open(CONFIG_FILE, 'r') as f: config = json.load(f) except: return False # Check if enabled if not config.get("enabled", False): return False # Get interval in seconds interval_minutes = config.get("interval_minutes", 5) interval_seconds = interval_minutes * 60 # Check last auto-check time last_check = config.get("last_check") if not last_check: # First run, trigger immediately return True # Parse last check time try: from datetime import datetime last_check_dt = datetime.fromisoformat(last_check) current_dt = datetime.now() elapsed = (current_dt - last_check_dt).total_seconds() # Trigger if enough time has passed return elapsed >= interval_seconds except: # If parsing fails, trigger to be safe return True # Main execution if should_trigger_auto_check(): # Create flag file for the AI to notice try: with open(FLAG_FILE, 'w') as f: f.write(f"Auto-check triggered at {time.strftime('%Y-%m-%d %H:%M:%S')}") # Also update timestamp to prevent immediate re-trigger # (The auto_process tool will update this properly, but this prevents double-trigger) with open(CONFIG_FILE, 'r') as f: config = json.load(f) config["last_check"] = time.strftime("%Y-%m-%dT%H:%M:%S") with open(CONFIG_FILE, 'w') as f: json.dump(config, f, indent=2) except Exception as e: # Silent failure - hooks shouldn't interrupt workflow pass # Always exit cleanly sys.exit(0)