Claude IPC MCP

Instructions

Implementation Reference

• src/claude_ipc_server.py:909-922 (schema)

  Input schema for the 'register' MCP tool, requiring an 'instance_id' parameter.
  Tool( name="register", description="Register this Claude instance with the IPC system", inputSchema={ "type": "object", "properties": { "instance_id": { "type": "string", "description": "Unique identifier for this instance (e.g., 'wsl1', 'wsl2')" } }, "required": ["instance_id"] } ),
• src/claude_ipc_server.py:1083-1105 (handler)

  MCP tool handler for 'register' that sends the registration request to the internal message broker and stores the session token.
  if name == "register": # Calculate auth token instance_id = arguments["instance_id"] shared_secret = os.environ.get("IPC_SHARED_SECRET", "") auth_token = "" if shared_secret: import hashlib auth_token = hashlib.sha256(f"{instance_id}:{shared_secret}".encode()).hexdigest() response = BrokerClient.send_request({ "action": "register", "instance_id": instance_id, "auth_token": auth_token }) # Store session token for this MCP instance if response.get("status") == "ok" and response.get("session_token"): # Store in a global variable for this MCP session global current_session_token, current_instance_id current_session_token = response["session_token"] current_instance_id = instance_id return [TextContent(type="text", text=json.dumps(response, indent=2))]
• src/claude_ipc_server.py:606-651 (helper)

  Core registration logic in the MessageBroker's _process_request method, which handles the 'register' action: validates ID and auth, generates session token, persists instance and session to DB.
  if action == "register": instance_id = request["instance_id"] # Validate instance ID format if not self._validate_instance_id(instance_id): return {"status": "error", "message": "Invalid instance ID format. Use 1-32 alphanumeric characters, hyphens, or underscores."} # Rate limit registration attempts (use IP or a special key) if not self.rate_limiter.is_allowed(f"register_{instance_id}"): return {"status": "error", "message": "Too many registration attempts. Please wait."} # Validate auth token (shared secret) auth_token = request.get("auth_token") shared_secret = os.environ.get("IPC_SHARED_SECRET", "") if shared_secret: import hashlib expected_token = hashlib.sha256(f"{instance_id}:{shared_secret}".encode()).hexdigest() if auth_token != expected_token: return {"status": "error", "message": "Invalid auth token"} # Generate session token session_token = secrets.token_urlsafe(32) # Register instance self.instances[instance_id] = datetime.now() # Save to database self._save_instance_to_db(instance_id) self._save_session_to_db(session_token, instance_id) # Preserve existing queue or create new one if instance_id not in self.queues: self.queues[instance_id] = [] queued_count = 0 else: queued_count = len(self.queues[instance_id]) response = { "status": "ok", "session_token": session_token, "message": f"Registered {instance_id}" } if queued_count > 0: response["message"] = f"Registered {instance_id} with {queued_count} queued messages" return response