mcp2mqtt

Instructions

Implementation Reference

• src/mcp2mqtt/server.py:290-292 (handler)

  Executes the 'get_pico_info' tool by preparing the 'INFO' message to be sent over MQTT to retrieve Pico device information.
  elif name == "get_pico_info": message = "INFO"
• src/mcp2mqtt/server.py:215-243 (registration)

  Registers and lists all available tools including 'get_pico_info' dynamically from the configuration, defining their schemas.
  @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:215-243 (schema)

  Dynamically generates the input schema for 'get_pico_info' (likely empty parameters) from the loaded configuration.
  @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:244-327 (handler)

  The main @server.call_tool() handler that dispatches to the specific logic for 'get_pico_info' and sends MQTT message.
  @server.call_tool() async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[types.TextContent | types.ImageContent]: """Handle tool execution requests.""" try: logger.info(f"Tool call received - Name: {name}, Arguments: {arguments}") # 检查工具是否存在 if name not in config.tools: return [types.TextContent( type="text", text=f"Error: Tool {name} not found" )] tool_config = config.tools[name] # 验证参数 if arguments is None: arguments = {} # 检查必需参数 for param in tool_config.parameters: if param.get('required', False) and param['name'] not in arguments: return [types.TextContent( type="text", text=f"Error: Missing required parameter {param['name']}" )] # 验证枚举值 if 'enum' in param and param['name'] in arguments: if arguments[param['name']] not in param['enum']: return [types.TextContent( type="text", text=f"Error: Invalid value for {param['name']}" )] # 准备消息 message = None 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}" elif name == "get_pico_info": message = "INFO" elif name == "led_control": state = arguments.get("state", "").lower() if state not in ["on", "off"]: return [types.TextContent( type="text", text="Error: State must be 'on' or 'off'" )] message = f"LED {state}" else: return [types.TextContent( type="text", text=f"Error: Unknown tool {name}" )] # 发送消息并等待响应 mqtt_connection = MQTTConnection(config) response = await mqtt_connection.connect_and_send( topic=tool_config.mqtt_topic, message=message, response_topic=tool_config.response_topic ) return [types.TextContent( type="text", text=response if response else f"{config.mqtt_response_start_string} {message} OK" )] except Exception as e: logger.error(f"Error handling tool call: {e}") return [types.TextContent( type="text", text=f"Error: {str(e)}" )]