Query topic

Instructions

Implementation Reference

• src/joulescope_mcp/service.py:439-448 (handler)

  Core handler logic for query_topic. Uses the driver session to select a device, constructs the full topic, performs driver.query(), and returns device_path, full topic, and JSON-serializable value.

  def query_topic(self, topic: str, device_path: str | None = None) -> dict[str, Any]:
      with self._driver_session() as driver:
          device = self._select_device(driver, device_path=device_path, require_js220=False)
          full_topic = self._full_topic(device, topic)
          driver.open(device, mode="restore")
          try:
              value = driver.query(full_topic)
          finally:
              driver.close(device)
      return {"device_path": device, "topic": full_topic, "value": _jsonable(value)}

• src/joulescope_mcp/server.py:191-201 (registration)

  MCP tool registration using @mcp.tool decorator with title 'Query topic', description, read_only annotations, and structured_output=True. Delegates to service.query_topic().

  @mcp.tool(
      title="Query topic",
      description="Query a JouleScope driver topic. Provide a relative topic such as c/fw/version or an absolute device topic.",
      annotations=read_only,
      structured_output=True,
  )
  def query_topic(topic: str, device_path: str | None = None) -> dict[str, Any]:
      try:
          return service.query_topic(topic=topic, device_path=device_path)
      except JoulescopeMcpError as exc:
          raise _tool_error(exc) from exc

• src/joulescope_mcp/service.py:473-485 (helper)

  Helper function _full_topic that resolves a relative topic (e.g., 'c/fw/version') to an absolute device topic (e.g., 'u/js220/005920/c/fw/version').

  def _full_topic(self, device: str, topic: str) -> str:
      topic = topic.strip()
      if not topic:
          raise JoulescopeMcpError("topic must not be empty")
      if topic.startswith(device + "/"):
          return topic
      if (
          topic.startswith(("u/", "s/", "c/", "h/"))
          and not topic.startswith(device)
          and topic.startswith(("u/js", "s/js"))
      ):
          return topic
      return f"{device}/{topic.lstrip('/')}"

• src/joulescope_mcp/service.py:65-76 (schema)

  Helper function _jsonable that converts driver/numpy values to JSON-serializable types, used to format the query_topic return value.

  def _jsonable(value: Any) -> Any:
      """Convert common driver/numpy values into JSON-serializable values."""
  
      if hasattr(value, "item"):
          return value.item()
      if isinstance(value, dict):
          return {str(k): _jsonable(v) for k, v in value.items()}
      if isinstance(value, (list, tuple)):
          return [_jsonable(v) for v in value]
      if isinstance(value, bytes):
          return value.hex()
      return value

• tests/test_service.py:194-199 (helper)

  Test confirming query_topic accepts relative topics and returns the correct full topic.

  def test_query_and_publish_topics_accept_relative_topics() -> None:
      driver = FakeDriver()
      svc = service_with(driver)
      assert svc.query_topic("c/fw/version")["topic"] == "u/js220/005920/c/fw/version"
      assert svc.publish_topic("s/led/en", 0)["topic"] == "u/js220/005920/s/led/en"
      assert ("u/js220/005920/s/led/en", 0) in driver.published