Claude IPC MCP

Instructions

Implementation Reference

• src/claude_ipc_server.py:923-948 (schema)

  Input schema for the 'send' MCP tool defining parameters from_id, to_id, content (required) and optional data.
  Tool( 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 tool handler for 'send': checks registration, builds message, sends to internal broker via BrokerClient, returns broker response.
  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:905-1078 (registration)

  Registration of 'send' tool via list_tools() returning Tool definitions including 'send'.
  @app.list_tools() async def list_tools() -> List[Tool]: """List available tools""" return [ 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"] } ), Tool( 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"] } ), Tool( name="broadcast", description="Broadcast a message to all other Claude instances", inputSchema={ "type": "object", "properties": { "from_id": { "type": "string", "description": "Your instance ID" }, "content": { "type": "string", "description": "Message content" }, "data": { "type": "object", "description": "Optional structured data" } }, "required": ["from_id", "content"] } ), Tool( name="check", description="Check for new messages", inputSchema={ "type": "object", "properties": { "instance_id": { "type": "string", "description": "Your instance ID" } }, "required": ["instance_id"] } ), Tool( name="list_instances", description="List all active Claude instances", inputSchema={ "type": "object", "properties": {} } ), Tool( name="share_file", description="Share file content with another instance", inputSchema={ "type": "object", "properties": { "from_id": { "type": "string", "description": "Your instance ID" }, "to_id": { "type": "string", "description": "Target instance ID" }, "filepath": { "type": "string", "description": "Path to file to share" }, "description": { "type": "string", "description": "Description of the file" } }, "required": ["from_id", "to_id", "filepath"] } ), Tool( name="share_command", description="Execute a command and share output with another instance", inputSchema={ "type": "object", "properties": { "from_id": { "type": "string", "description": "Your instance ID" }, "to_id": { "type": "string", "description": "Target instance ID" }, "command": { "type": "string", "description": "Command to execute" }, "description": { "type": "string", "description": "Description of what this command does" } }, "required": ["from_id", "to_id", "command"] } ), Tool( name="rename", description="Rename your instance ID (rate limited to once per hour)", inputSchema={ "type": "object", "properties": { "old_id": { "type": "string", "description": "Your current instance ID" }, "new_id": { "type": "string", "description": "The new instance ID you want" } }, "required": ["old_id", "new_id"] } ), Tool( 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:653-712 (handler)

  Core implementation of 'send' action in MessageBroker._process_request: handles validation, large message storage, name resolution/forwarding, queuing, persistence to SQLite, and response.
  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.send_request: utility to send JSON requests to the TCP message broker socket and return response.
  @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}"}