led

Instructions

Implementation Reference

• src/mcp2tcp/server.py:57-107 (registration)

  Dynamically registers the 'led' tool if config.yaml defines a command starting with 'CMD_LED'. Generates tool name 'led' and inputSchema from command parameters and placeholders.

  @server.list_tools()
  async def handle_list_tools() -> List[types.Tool]:
      """List available tools."""
      if config is None:
          return []
          
      tools = []
      for cmd_id, command in config.commands.items():
          # 从命令字符串中提取命令名（去掉CMD_前缀）
          cmd_name = command.command.split()[0].replace("CMD_", "").lower()
          
          # 构建参数描述
          properties = {}
          required = []
          
          # 使用配置文件中的参数定义
          if hasattr(command, 'parameters'):
              for param in command.parameters:
                  properties[param['name']] = {
                      "type": param['type'],
                      "description": param['description'],
                      **({"enum": param['enum']} if 'enum' in param else {})
                  }
                  if param.get('required', False):
                      required.append(param['name'])
          else:
              # 如果没有参数定义，从命令字符串中提取
              import re
              param_names = re.findall(r'\{(\w+)\}', command.command)
              for param_name in param_names:
                  properties[param_name] = {
                      "type": "string",
                      "description": f"Parameter {param_name} for the {cmd_name} command",
                      "examples": [p.format(**{param_name: "value"}) for p in command.prompts if "{" + param_name + "}" in p]
                  }
                  required.append(param_name)
  
          tool = types.Tool(
              name=cmd_name,  # 使用命令名作为工具名
              description=command.prompts[0] if command.prompts else f"Execute {cmd_name} command",
              inputSchema={
                  "type": "object",
                  "properties": properties,
                  "required": required,
                  "additionalProperties": False
              }
          )
          tools.append(tool)
          logger.debug(f"Registered tool: {cmd_name} with parameters: {properties}")
      
      return tools

• src/mcp2tcp/server.py:109-192 (handler)

  Generic handler for all tools including 'led': matches tool name to config command, validates arguments, formats the command string (e.g., 'CMD_LED {state}'), sends via TCP, returns response.

  @server.call_tool()
  async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[types.TextContent]:
      """Handle tool execution requests."""
      if config is None:
          return [types.TextContent(
              type="text",
              text="Error: Configuration not loaded"
          )]
          
      try:
          logger.info(f"Tool call received - Name: {name}, Arguments: {arguments}")
          
          # 查找对应的命令
          cmd_found = None
          for cmd_id, command in config.commands.items():
              cmd_name = command.command.split()[0].replace("CMD_", "").lower()
              if cmd_name == name:
                  cmd_found = command
                  break
          
          if cmd_found is None:
              error_msg = f"Error: Unknown tool '{name}'\n"
              error_msg += "Please check:\n"
              error_msg += "1. Tool name is correct\n"
              error_msg += "2. Tool is configured in config.yaml"
              logger.error(error_msg)
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
  
          if arguments is None:
              arguments = {}
          
          # 验证必需的参数
          import re
          param_names = re.findall(r'\{(\w+)\}', cmd_found.command)
          missing_params = [param for param in param_names if param not in arguments]
          if missing_params:
              error_msg = f"Error: Missing required parameters: {', '.join(missing_params)}\n"
              error_msg += "Please provide all required parameters."
              logger.error(error_msg)
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
          
          # 发送命令并等待响应
          try:
              response = tcp_connection.send_command(cmd_found, arguments)
              logger.debug(f"Command response: {response}")
              return response
          except ConnectionError as e:
              error_msg = f"Error: Connection failed - {str(e)}\n"
              error_msg += "Please check:\n"
              error_msg += "1. TCP server is running\n"
              error_msg += "2. Connection settings are correct"
              logger.error(error_msg)
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
          except TimeoutError as e:
              error_msg = f"Error: Command timeout - {str(e)}\n"
              error_msg += "Please check:\n"
              error_msg += "1. Device is responding\n"
              error_msg += "2. Timeout settings are appropriate"
              logger.error(error_msg)
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
          
      except Exception as e:
          error_msg = f"Error: {str(e)}\n"
          error_msg += "Please check:\n"
          error_msg += "1. Configuration is correct\n"
          error_msg += "2. Device is functioning properly"
          logger.error(f"Unexpected error: {str(e)}", exc_info=True)
          return [types.TextContent(
              type="text",
              text=error_msg
          )]

• src/mcp2tcp/server.py:284-360 (helper)

  Core helper that executes the formatted 'CMD_LED {state}' command over TCP socket, handles connection, sending with \r\n, receiving until timeout or end marker, formats as TextContent.

  def send_command(self, command: Command, arguments: dict[str, Any] | None) -> list[types.TextContent]:
      """Send a command to the TCP server and return result according to MCP protocol."""
      try:
          if not self.socket:
              if not self.connect():
                  return [types.TextContent(
                      type="text",
                      text=f"Failed to connect to TCP server at {self.remote_ip}:{self.port}"
                  )]
          if command.data_type == "ascii":
              # 准备命令
              cmd_str = command.command.format(**arguments)
              # 确保命令以\r\n结尾
              cmd_str = cmd_str.rstrip() + '\r\n'  # 移除可能的空白字符，强制添加\r\n
              command_bytes = cmd_str.encode()
              logger.info(f"Sending command: {cmd_str.strip()}")
              logger.info(f"Sent command: {command_bytes.strip().decode('ascii')}")
              self.socket.sendall(command_bytes)
  
          elif command.data_type == "hex":
              command_bytes = bytes.fromhex(command.command.replace(" ", ""))
              logger.info(f"Sent command: {command.command}")
              self.socket.sendall(command_bytes)
          self.socket.settimeout(self.receive_timeout)
          responses = []
          while True:
              try:
                  response = self.socket.recv(4096)
                  if response:
                      logger.debug(f"Received data: {response}")
                      responses.append(response)
                      if command.data_type == "ascii" and response.endswith(b'\r\n'):
                          break
                      elif command.data_type == "hex":
                          break
                  else:
                      break
              except socket.timeout as e:
                  logger.error(f"TCP receive timeout: {str(e)}")
                  return [types.TextContent(
                      type="text",
                      text=f"TCP receive timeout: {str(e)}"
                  )]
  
          if not responses:
              return [types.TextContent(
                  type="text",
                  text=f"No response received from TCP server within {self.receive_timeout} seconds"
              )]
  
          first_response = responses[0].decode().strip()
          logger.info(f"Received response: {first_response}")
  
          if self.response_start_string in first_response:
              return [types.TextContent(
                  type="text",
                  text=first_response
              )]
          else:
              return [types.TextContent(
                  type="text",
                  text=f"Invalid response: {first_response}"
              )]
  
      except socket.timeout as e:
          logger.error(f"TCP send timeout: {str(e)}")
          return [types.TextContent(
              type="text",
              text=f"TCP send timeout: {str(e)}"
          )]
      except Exception as e:
          logger.error(f"TCP error: {str(e)}")
          return [types.TextContent(
              type="text",
              text=f"TCP error: {str(e)}"
          )]

• tests/tcp_server.py:38-42 (helper)

  Mock device-side LED handler used in tests, simulates the TCP server response for CMD_LED command.

  def set_led(self, state):
      if state.lower() in ["on", "off"]:
          self.led_state = state.lower()
          return f"CMD LED state set to {state}\r\n"
      return f"CMD Error: Invalid LED state. Use 'on' or 'off'\r\n"

• tests/tcp_server.py:60-66 (handler)

  Test TCP server handler that processes 'CMD_LED {state}' and calls set_led mock implementation.

  # 处理LED控制命令
  elif command.startswith("CMD_LED"):
      try:
          state = command.split()[1]
          return device_state.set_led(state)
      except IndexError:
          return "CMD Error: Missing LED state parameter\r\n"