execute_service_command
Execute a command on any CloudNet service console by providing the service identifier and command string. Automate service management tasks.
Instructions
Executes the specified command on a service console
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| identifier | Yes | The name or unique id of the service | |
| command | Yes | The command to execute on the service |
Implementation Reference
- src/cloudnet_mcp/server.py:150-161 (schema)Tool schema/registration for execute_service_command — defines name, description, and inputSchema with identifier and command parameters.
types.Tool( name="execute_service_command", description="Executes the specified command on a service console", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the service"}, "command": {"type": "string", "description": "The command to execute on the service"} }, "required": ["identifier", "command"], }, ), - src/cloudnet_mcp/server.py:212-216 (handler)Handler implementation for execute_service_command — extracts identifier and command from arguments, sends a POST request to the CloudNet API at service/{identifier}/command.
elif name == "execute_service_command": identifier = arguments.get("identifier") cmd = arguments.get("command") data = await client.request("POST", f"service/{identifier}/command", json={"command": cmd}) return [types.TextContent(type="text", text=str(data))] - src/cloudnet_mcp/server.py:60-162 (registration)Tool listing/registration via @app.list_tools() — all tools including execute_service_command are declared here.
@app.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="get_nodes", description="List all nodes in the CloudNet cluster", inputSchema={ "type": "object", "properties": {}, }, ), types.Tool( name="get_node_info", description="Get detailed information about a specific node", inputSchema={ "type": "object", "properties": { "node_id": {"type": "string", "description": "The ID of the node"} }, "required": ["node_id"], }, ), types.Tool( name="get_services", description="List all smart services", inputSchema={ "type": "object", "properties": {}, }, ), types.Tool( name="get_online_players", description="Get a list of online players based on the query parameters", inputSchema={ "type": "object", "properties": { "limit": {"type": "integer", "description": "The maximum amount of players to respond with"}, "skip": {"type": "integer", "description": "The amount of players to skip"}, "sort": {"type": "string", "enum": ["asc", "desc"], "description": "Sort players by name"}, }, }, ), types.Tool( name="get_player_info", description="Get a player by their unique id or name", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the player"} }, "required": ["identifier"], }, ), types.Tool( name="kick_player", description="Kicks a given player from the entire network", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the player"}, "message": {"type": "string", "description": "The kick message/reason"} }, "required": ["identifier", "message"], }, ), types.Tool( name="send_player_message", description="Sends a chat message to a given player", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the player"}, "message": {"type": "string", "description": "The chat message to send"} }, "required": ["identifier", "message"], }, ), types.Tool( name="execute_player_command", description="Executes a command for a given player", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the player"}, "command": {"type": "string", "description": "The command to execute (without prefixing slash)"}, "redirectToServer": {"type": "boolean", "description": "Redirect downstream if not found on proxy"} }, "required": ["identifier", "command"], }, ), types.Tool( name="execute_service_command", description="Executes the specified command on a service console", inputSchema={ "type": "object", "properties": { "identifier": {"type": "string", "description": "The name or unique id of the service"}, "command": {"type": "string", "description": "The command to execute on the service"} }, "required": ["identifier", "command"], }, ), ] - src/cloudnet_mcp/server.py:16-56 (helper)CloudNetClient helper class — handles authentication, token refresh, and HTTP requests used by the handler.
class CloudNetClient: def __init__(self, base_url: str, user: str, password: str): self.base_url = base_url.rstrip("/") self.user = user self.password = password self.token = None self.client = httpx.AsyncClient() async def _authenticate(self): resp = await self.client.post( f"{self.base_url}/auth", auth=(self.user, self.password) ) resp.raise_for_status() data = resp.json() self.token = data.get("token") self.client.headers.update({"Authorization": f"Bearer {self.token}"}) async def request(self, method: str, endpoint: str, **kwargs): if not self.token: await self._authenticate() path = endpoint.lstrip("/") url = f"{self.base_url}/{path}" try: resp = await self.client.request(method, url, **kwargs) if resp.status_code == 401: # Token might be expired, re-authenticate and retry await self._authenticate() resp = await self.client.request(method, url, **kwargs) resp.raise_for_status() if resp.status_code == 204: return {"status": "success"} return resp.json() except httpx.HTTPError as e: return {"error": str(e)} async def close(self): await self.client.aclose()