list_homes

Instructions

Args:
    refresh: 是否强制刷新缓存

Implementation Reference

• mcp_server/server.py:132-142 (handler)

  MCP tool handler for 'list_homes' - calls the backend API endpoint /homes/ with optional refresh parameter.

  @mcp.tool()
  async def list_homes(refresh: bool = False) -> dict:
      """列出所有家庭及其设备概览。
  
      Args:
          refresh: 是否强制刷新缓存
      """
      params = {}
      if refresh:
          params["refresh"] = "true"
      return await _request("GET", "/homes/", params=params)

• app/api/homes.py:11-37 (schema)

  Flask API endpoint handler for listing homes - validates the refresh query parameter and calls HomeService.list_homes.

  @homes_ns.route("/", methods=["GET"])
  @auth_required
  @limiter.limit("60 per minute")
  def list_homes():
      """获取家庭列表
      ---
      tags:
        - 家庭
      security:
        - cookieAuth: []
        - bearerAuth: []
      parameters:
        - in: query
          name: refresh
          type: boolean
          default: false
      responses:
        200:
          description: 家庭列表
      """
      refresh = request.args.get("refresh", "false").lower() == "true"
      try:
          homes = HomeService.list_homes(get_current_user_id(), refresh=refresh)
          return success(homes)
      except Exception as e:
          return error(str(e), 500)

• app/services/home_service.py:10-25 (helper)

  Core business logic for list_homes - returns cached homes or refreshes from the Mi Jia API if refresh is requested or cache is empty.

  def list_homes(user_id: int, refresh: bool = False) -> list[dict]:
      if refresh:
          return HomeService._refresh_homes(user_id)
  
      cached = HomeCache.query.filter_by(user_id=user_id).all()
      if not cached:
          return HomeService._refresh_homes(user_id)
  
      return [
          {
              "home_id": h.home_id,
              "name": h.name,
              "room_list": h.room_list,
          }
          for h in cached
      ]

• mcp_server/server.py:32-132 (registration)

  The @mcp.tool() decorator registers 'list_homes' as an MCP tool in the FastMCP server.

  @mcp.tool()
  async def list_devices(home_id: str | None = None, refresh: bool = False) -> dict:
      """列出所有米家智能设备。返回设备ID、名称、型号、在线状态等信息。
  
      Args:
          home_id: 按家庭ID过滤，不传则返回全部设备
          refresh: 是否强制刷新设备列表缓存
      """
      params = {}
      if home_id:
          params["home_id"] = home_id
      if refresh:
          params["refresh"] = "true"
      return await _request("GET", "/devices/", params=params)
  
  
  @mcp.tool()
  async def get_device(did: str) -> dict:
      """获取设备详情，包含设备规格、可用属性列表和动作列表。
  
      Args:
          did: 设备ID
      """
      return await _request("GET", f"/devices/{quote(did)}")
  
  
  # ── 设备属性读写 ──
  
  
  @mcp.tool()
  async def get_property(did: str, prop_name: str) -> dict:
      """读取设备属性值，例如灯光亮度、空调温度、开关状态等。
  
      Args:
          did: 设备ID
          prop_name: 属性名称，如 power、brightness、temperature
      """
      return await _request("GET", f"/devices/{quote(did)}/props/{prop_name}")
  
  
  @mcp.tool()
  async def set_property(did: str, prop_name: str, value) -> dict:
      """设置设备属性值，用于控制设备。例如开灯、调亮度、设温度等。
  
      Args:
          did: 设备ID
          prop_name: 属性名称，如 power、brightness、temperature
          value: 属性值，类型取决于属性定义。常见值：power 为 "on"/"off"，brightness 为 0-100，temperature 为数字
      """
      return await _request("PUT", f"/devices/{quote(did)}/props/{prop_name}", json_data={"value": value})
  
  
  # ── 设备动作 ──
  
  
  @mcp.tool()
  async def run_action(did: str, action_name: str, value: dict | None = None) -> dict:
      """执行设备动作，例如扫地机开始清扫、播放音乐等。
  
      Args:
          did: 设备ID
          action_name: 动作名称，如 start-sweep、stop-sweeping
          value: 动作参数，可选
      """
      body = {"value": value} if value is not None else {}
      return await _request("POST", f"/devices/{quote(did)}/actions/{action_name}", json_data=body)
  
  
  # ── 场景管理 ──
  
  
  @mcp.tool()
  async def list_scenes(home_id: str | None = None, refresh: bool = False) -> dict:
      """列出所有米家场景。
  
      Args:
          home_id: 按家庭ID过滤
          refresh: 是否强制刷新缓存
      """
      params = {}
      if home_id:
          params["home_id"] = home_id
      if refresh:
          params["refresh"] = "true"
      return await _request("GET", "/scenes/", params=params)
  
  
  @mcp.tool()
  async def run_scene(scene_id: str) -> dict:
      """执行一个米家场景，触发该场景中预设的所有设备操作。
  
      Args:
          scene_id: 场景ID
      """
      return await _request("POST", f"/scenes/{scene_id}/run")
  
  
  # ── 家庭管理 ──
  
  
  @mcp.tool()