Ecovacs MCP Server

Instructions

Implementation Reference

• ecovacs_mcp/robot_mcp_stdio.py:74-89 (handler)

  The handler function for the 'set_charging' tool, decorated with @mcp.tool() for registration. It takes nickname and act parameters, validates with Pydantic Fields (schema), and calls the shared API helper with 'Charge' command.

  @mcp.tool()
  async def set_charging(
      nickname: str = Field(description="Robot nickname, used to find device"),
      act: str = Field(description="Robot action, go-start begin returning to charging station, stopGo stop returning to charging station", default="go-start")
  ) -> dict:
      """
      Make robot return to charging station
      
      Args:
          nickname: Robot nickname, used to find device
          act: Robot action, go-start begin returning to charging station, stopGo stop returning to charging station
      
      Returns:
          Dict: Dictionary containing execution results
      """
      return await call_api(ENDPOINT_ROBOT_CTL, {"nickName": nickname, "cmd": "Charge", "act": act})

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

  Shared helper function used by set_charging to make HTTP API calls to the Ecovacs API.

  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": []}