kick_mqtt_client
Disconnect a client from the MQTT broker using its client ID to manage connections and enforce access control.
Instructions
Disconnect a client from the MQTT broker by client ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| request | Yes |
Implementation Reference
- MCP tool handler for 'kick_mqtt_client'. Validates input, logs, delegates to EMQXClient.kick_client, and returns result. Also serves as registration via @mcp.tool decorator.
@mcp.tool(name="kick_mqtt_client", description="Disconnect a client from the MQTT broker by client ID") async def kick_client(request): """Handle kick client request Args: request: MCP request containing client identifier - clientid: Client ID (required) - The unique identifier of the client to disconnect Returns: MCPResponse: Response object with the result of the disconnect operation """ self.logger.info("Handling kick client request") # Extract required client ID parameter clientid = request.get("clientid") if not clientid: self.logger.error("Client ID is required but was not provided") return {"error": "Client ID is required"} # Kick client from EMQX result = await self.emqx_client.kick_client(clientid) self.logger.info(f"Client '{clientid}' disconnect request processed") return result - Core helper method in EMQXClient that executes the HTTP DELETE request to /clients/{clientid} to disconnect the MQTT client.
async def kick_client(self, clientid: str): """ Kick out (disconnect) a client from the MQTT broker. Uses the EMQX HTTP API to forcibly disconnect a client identified by its client ID. Args: clientid (str): The unique identifier of the client to disconnect Returns: dict: Response from the EMQX API or error information """ url = f"{self.api_url}/clients/{clientid}" self.logger.info(f"Kicking out client with ID: {clientid}") async with httpx.AsyncClient() as client: try: response = await client.delete( url, headers=self._get_auth_header(), timeout=30 ) response.raise_for_status() # For successful delete operations, return a success message if response.status_code == 204: # No Content return {"success": True, "message": f"Client {clientid} has been disconnected"} return self._handle_response(response) except Exception as e: self.logger.error(f"Error kicking out client {clientid}: {str(e)}") return {"error": str(e)}