Claude IPC MCP

Instructions

Implementation Reference

• src/claude_ipc_server.py:924-948 (registration)

  MCP tool registration for 'send', including name, description, and input schema.
  name="send", description="Send a message to another Claude instance", inputSchema={ "type": "object", "properties": { "from_id": { "type": "string", "description": "Your instance ID" }, "to_id": { "type": "string", "description": "Target instance ID" }, "content": { "type": "string", "description": "Message content" }, "data": { "type": "object", "description": "Optional structured data to send" } }, "required": ["from_id", "to_id", "content"] } ),
• src/claude_ipc_server.py:1107-1122 (handler)

  MCP handler for the 'send' tool: constructs message and forwards to internal IPC broker via BrokerClient.send_request.
  elif name == "send": if not current_session_token: return [TextContent(type="text", text="Error: Not registered. Please register first.")] message = { "content": arguments["content"], "data": arguments.get("data", {}) } response = BrokerClient.send_request({ "action": "send", "from_id": arguments["from_id"], "to_id": arguments["to_id"], "message": message, "session_token": current_session_token }) return [TextContent(type="text", text=json.dumps(response, indent=2))]
• src/claude_ipc_server.py:653-712 (handler)

  Core implementation of 'send' action in MessageBroker: handles validation, large message persistence, name resolution/forwarding, queuing, and database persistence.
  elif action == "send": from_id = request["from_id"] to_id = request["to_id"] message = request["message"] # Validate to_id format if not self._validate_instance_id(to_id): return {"status": "error", "message": "Invalid recipient ID format"} # Check message size (10KB threshold) content = message.get("content", "") content_size = len(content.encode('utf-8')) size_threshold = 10 * 1024 # 10KB if content_size > size_threshold: # Save large message to file filepath = self._save_large_message(from_id, to_id, content) if filepath: # Create summary and update message summary = self._create_summary(content) message = { "content": f"{summary} Full content saved to: {filepath}", "data": message.get("data", {}) } message["data"]["large_message_file"] = filepath message["data"]["original_size_kb"] = round(content_size / 1024, 1) # Resolve name through forwarding if needed resolved_to = self._resolve_name(to_id) forwarded = resolved_to != to_id # Create queue for future instances if it doesn't exist if resolved_to not in self.queues: self.queues[resolved_to] = [] future_delivery = True else: future_delivery = not (resolved_to in self.instances) # Check queue limit (100 messages per instance) if len(self.queues[resolved_to]) >= 100: return {"status": "error", "message": f"Message queue full for {resolved_to} (100 message limit)"} msg_data = { "from": from_id, "to": resolved_to, "timestamp": datetime.now().isoformat(), "message": message } self.queues[resolved_to].append(msg_data) # Save to SQLite self._save_message_to_db(from_id, resolved_to, msg_data) if forwarded: return {"status": "ok", "message": f"Message forwarded from {to_id} to {resolved_to}"} elif future_delivery: return {"status": "ok", "message": f"Message queued for {resolved_to} (not yet registered)"} else: return {"status": "ok", "message": "Message sent"}
• src/claude_ipc_server.py:858-875 (helper)

  BrokerClient helper method used by MCP tools to communicate with the internal TCP message broker server.
  @staticmethod def send_request(request: Dict[str, Any]) -> Dict[str, Any]: """Send a request to the broker""" try: client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.settimeout(5.0) client_socket.connect((IPC_HOST, IPC_PORT)) client_socket.send(json.dumps(request).encode('utf-8')) response_data = client_socket.recv(65536).decode('utf-8') response = json.loads(response_data) client_socket.close() return response except Exception as e: return {"status": "error", "message": f"Broker connection failed: {e}"}