Ecovacs MCP Server

Instructions

Implementation Reference

• ecovacs_mcp/robot_mcp_stdio.py:58-58 (registration)

  Registers the set_cleaning tool using the @mcp.tool() decorator.

  @mcp.tool()

• ecovacs_mcp/robot_mcp_stdio.py:59-73 (handler)

  The main handler function implementing the set_cleaning tool logic. It calls the API with the 'Clean' command and the specified action (start, resume, pause, stop).

  async def set_cleaning(
      nickname: str = Field(description="Robot nickname, supports fuzzy matching", default=""),
      act: str = Field(description="Cleaning action, s-start cleaning, r-resume cleaning, p-pause cleaning, h-stop cleaning", default="s")
  ) -> dict:
      """
      Start robot cleaning
      
      Args:
          nickname: Robot nickname, used to find device
          act: Cleaning action s-start cleaning, r-resume cleaning, p-pause cleaning, h-stop cleaning
      Returns:
          Dict: Dictionary containing execution results
      """
      return await call_api(ENDPOINT_ROBOT_CTL, {"nickName": nickname, "cmd": "Clean", "act": act})

• ecovacs_mcp/robot_mcp_stdio.py:60-61 (schema)

  Input schema definitions using Pydantic Field for the nickname and act parameters.

  nickname: str = Field(description="Robot nickname, supports fuzzy matching", default=""),
  act: str = Field(description="Cleaning action, s-start cleaning, r-resume cleaning, p-pause cleaning, h-stop cleaning", default="s")

• ecovacs_mcp/robot_mcp_stdio.py:21-56 (helper)

  Helper function for making API calls to the Ecovacs service, used by set_cleaning.

  async def call_api(endpoint: str, params: dict, method: str = 'post') -> dict:
      """
      General API call function
      
      Args:
          endpoint: API endpoint
          params: Request parameters
          method: Request method, 'get' or 'post'
      
      Returns:
          Dict: API response result, format {"msg": "OK", "code": 0, "data": [...]}
      """
      # Build complete URL
      url = f"{API_URL}/{endpoint}"
      
      # Ensure all parameters are strings
      params = {k: str(v) for k, v in params.items()}
      
      # Add API key
      if API_KEY:
          params.update({"ak": API_KEY})
      
      try:
          async with httpx.AsyncClient() as client:
              headers = {"Content-Type": "application/json"}
              if method.lower() == 'get':
                  response = await client.get(url, params=params, timeout=REQUEST_TIMEOUT)
              else:
                  response = await client.post(url, json=params, headers=headers, timeout=REQUEST_TIMEOUT)
              
              response.raise_for_status()
              return response.json()
      
      except Exception as e:
          # Return unified error format when an error occurs
          return {"msg": f"Request failed: {str(e)}", "code": -1, "data": []}

• ecovacs_mcp/robot_mcp_stdio.py:17-17 (helper)

  Constant defining the API endpoint for robot control operations.

  ENDPOINT_ROBOT_CTL = "robot/ctl"