mcp2mqtt

Instructions

Implementation Reference

• src/mcp2mqtt/server.py:281-289 (handler)

  Executes the set_pwm tool: validates frequency argument (0-100), constructs MQTT payload 'PWM {frequency}'.

  if name == "set_pwm":
      frequency = arguments.get("frequency", 0)
      if not (0 <= frequency <= 100):
          return [types.TextContent(
              type="text",
              text="Error: Frequency must be between 0 and 100"
          )]
      message = f"PWM {frequency}"

• src/mcp2mqtt/server.py:215-242 (registration)

  Dynamically lists and registers MCP tools loaded from config.yaml, including set_pwm if configured, with schema derived from config parameters.

  @server.list_tools()
  async def handle_list_tools() -> List[types.Tool]:
      """List available tools."""
      tools = []
      for tool_name, tool_config in config.tools.items():
          tools.append(
              types.Tool(
                  name=tool_config.name,
                  description=tool_config.description,
                  inputSchema={
                      "type": "object",
                      "properties": {
                          param["name"]: {
                              "type": param["type"],
                              "description": param["description"],
                              **({"enum": param["enum"]} if "enum" in param else {})
                          }
                          for param in tool_config.parameters
                      },
                      "required": [
                          param["name"]
                          for param in tool_config.parameters
                          if param.get("required", False)
                      ]
                  }
              )
          )
      return tools

• src/mcp2mqtt/server.py:224-239 (schema)

  Constructs JSON schema for tool inputs from configuration parameters (used for set_pwm schema).

  inputSchema={
      "type": "object",
      "properties": {
          param["name"]: {
              "type": param["type"],
              "description": param["description"],
              **({"enum": param["enum"]} if "enum" in param else {})
          }
          for param in tool_config.parameters
      },
      "required": [
          param["name"]
          for param in tool_config.parameters
          if param.get("required", False)
      ]
  }