List topics

Instructions

Implementation Reference

• src/joulescope_mcp/service.py:461-471 (handler)

  The core implementation of list_topics. Calls self.device_info() to get all retained topics and their values, then optionally enriches each with metadata. Returns a dict with device_path and a list of topic items.

  def list_topics(self, device_path: str | None = None, include_metadata: bool = True) -> dict[str, Any]:
      info = self.device_info(device_path=device_path, include_metadata=include_metadata)
      topics = []
      values = info["values"]
      metadata = info.get("metadata") or {}
      for topic, value in values.items():
          item = {"topic": topic, "value": value}
          if include_metadata and topic in metadata:
              item["metadata"] = metadata[topic]
          topics.append(item)
      return {"device_path": info["device_path"], "topics": topics}

• src/joulescope_mcp/server.py:179-184 (schema)

  The @mcp.tool decorator registering list_topics as an MCP tool with title, description, read_only annotation, and structured_output=True. Defines the parameters device_path (optional str) and include_metadata (bool, default True).

  @mcp.tool(
      title="List topics",
      description="List retained JouleScope driver topics with current values and optional metadata.",
      annotations=read_only,
      structured_output=True,
  )

• src/joulescope_mcp/server.py:185-189 (handler)

  The MCP tool handler in the server layer. Delegates to service.list_topics() and wraps JoulescopeMcpError into ToolError.

  def list_topics(device_path: str | None = None, include_metadata: bool = True) -> dict[str, Any]:
      try:
          return service.list_topics(device_path=device_path, include_metadata=include_metadata)
      except JoulescopeMcpError as exc:
          raise _tool_error(exc) from exc

• src/joulescope_mcp/service.py:167-197 (helper)

  The device_info helper method called by list_topics. Subscribes to pub_retain and metadata_rsp_retain topics, collects values, and returns them with device_path.

  def device_info(self, device_path: str | None = None, include_metadata: bool = False) -> dict[str, Any]:
      values: dict[str, Any] = {}
      metadata: dict[str, Any] = {}
  
      def on_pub(topic: str, value: Any) -> None:
          values[topic] = _jsonable(value)
  
      def on_metadata(topic: str, value: Any) -> None:
          metadata[topic[:-1] if topic.endswith("$") else topic] = _jsonable(value)
  
      with self._driver_session() as driver:
          device = self._select_device(driver, device_path=device_path, require_js220=False)
          driver.open(device, mode="restore")
          try:
              driver.subscribe(device, "pub_retain", on_pub)
              driver.unsubscribe(device, on_pub)
              if include_metadata:
                  driver.subscribe(device, "metadata_rsp_retain", on_metadata)
                  driver.unsubscribe(device, on_metadata)
          finally:
              driver.close(device)
  
      def strip_prefix(items: dict[str, Any]) -> dict[str, Any]:
          prefix = device + "/"
          return {k[len(prefix) :] if k.startswith(prefix) else k: v for k, v in sorted(items.items())}
  
      return {
          "device_path": device,
          "values": strip_prefix(values),
          "metadata": strip_prefix(metadata) if include_metadata else None,
      }