Publish topic

Instructions

Implementation Reference

• src/joulescope_mcp/service.py:450-459 (handler)

  The actual implementation of publish_topic in Js220Service. Opens a driver session, selects the device, expands the topic via _full_topic, opens the device in restore mode, calls driver.publish() to send the value, then returns device path, full topic, and value.

  def publish_topic(self, topic: str, value: Any, 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:
              driver.publish(full_topic, value)
          finally:
              driver.close(device)
      return {"device_path": device, "topic": full_topic, "value": _jsonable(value)}

• src/joulescope_mcp/server.py:212-216 (handler)

  The MCP tool handler/decorator for publish_topic. It is decorated with @mcp.tool (title='Publish topic', annotations=write_tool, structured_output=True) and delegates to service.publish_topic, wrapping JoulescopeMcpError as ToolError.

  def publish_topic(topic: str, value: Any, device_path: str | None = None) -> dict[str, Any]:
      try:
          return service.publish_topic(topic=topic, value=value, device_path=device_path)
      except JoulescopeMcpError as exc:
          raise _tool_error(exc) from exc

• src/joulescope_mcp/server.py:203-211 (registration)

  The MCP tool registration: @mcp.tool decorator with title, description, destructive annotations (write_tool), and structured_output=True.

  @mcp.tool(
      title="Publish topic",
      description=(
          "Publish a value to a JouleScope driver topic. This exposes advanced JS220 capabilities; "
          "only use when you know the topic semantics."
      ),
      annotations=write_tool,
      structured_output=True,
  )

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

  Helper used within publish_topic (and query_topic) to resolve relative topics to fully-qualified device topic paths by prepending the device path if necessary.

  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('/')}"