MCP2Serial

Instructions

Implementation Reference

• src/mcp2serial/server.py:365-398 (handler)

  MCP tool call handler that dispatches execution for 'get_pico_info' by retrieving the command from config and sending it over serial via send_command.

  @server.call_tool()
  async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]:
      """Handle tool execution requests according to MCP protocol."""
      logger.info(f"Tool call received - Name: {name}, Arguments: {arguments}")
      
      try:
          if name not in config.commands:
              error_msg = f"[MCP2Serial v{VERSION}] 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"
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
  
          command = config.commands[name]
          if arguments is None:
              arguments = {}
          
          # 发送命令并返回 MCP 格式的响应
          return serial_connection.send_command(command, arguments)
  
      except Exception as e:
          logger.error(f"Error handling tool call: {str(e)}")
          error_msg = f"[MCP2Serial v{VERSION}] Error: {str(e)}\n"
          error_msg += "Please check:\n"
          error_msg += "1. Configuration is correct\n"
          error_msg += "2. Device is functioning properly"
          return [types.TextContent(
              type="text",
              text=error_msg
          )]

• src/mcp2serial/server.py:340-363 (registration)

  Registers the 'get_pico_info' tool dynamically from config.commands, including schema generation based on command template parameters.

  @server.list_tools()
  async def handle_list_tools() -> list[types.Tool]:
      """List available tools for the MCP service."""
      logger.info("Listing available tools")
      tools = []
      
      for cmd_id, command in config.commands.items():
          # 从命令字符串中提取参数名
          import re
          param_names = re.findall(r'\{(\w+)\}', command.command)
          properties = {name: {"type": "string"} for name in param_names}
          
          tools.append(types.Tool(
              name=cmd_id,
              description=f"Execute {cmd_id} command",
              inputSchema={
                  "type": "object",
                  "properties": properties,
                  "required": param_names
              },
              prompts=command.prompts
          ))
      
      return tools

• firmware/src/main.py:66-87 (handler)

  Device-side handler function that gathers and returns Pico board information (board name, MicroPython version, frequency, memory, disk usage). Called in response to 'PICO_INFO' serial command.

  def get_pico_info():
      d = uos.uname()
      board_name = d[4]
      micropython_version = d[2]
  
      system_freq = machine.freq() // 1000000  # 系统频率 (MHz)
  
      memory_info = gc.mem_free() + gc.mem_alloc()  # 内存信息
  
      disk_info = uos.statvfs('/')
      total_disk_size = disk_info[0] * disk_info[2]
      free_disk_size = disk_info[0] * disk_info[3]
  
      # 拼接信息字符串
      info = (
          f"Board: {board_name}, "
          f"MicroPython: {micropython_version}, "
          f"Freq: {system_freq} MHz, "
          f"Memory: {memory_info} bytes, "
          f"Disk: Total {total_disk_size} bytes, Free {free_disk_size} bytes"
      )
      return info

• firmware/src/main.py:107-109 (registration)

  Firmware serial input handler that recognizes 'PICO_INFO' command and invokes get_pico_info() to produce the response sent back over serial.

  elif user_input.strip() == "PICO_INFO":
      info = get_pico_info()
      print(f"OK {info}")

• src/mcp2serial/server.py:206-337 (helper)

  Helper method in SerialConnection that formats and sends the serial command for 'get_pico_info' (e.g., 'PICO_INFO'), reads the 'OK info' response from device, and returns it as MCP TextContent.

  def send_command(self, command: Command, arguments: Dict[str, Any]) -> list[types.TextContent]:
      """Send a command to the serial port and return result according to MCP protocol."""
      try:
          # 确保连接
          if not self.is_loopback and (not self.serial_port or not self.serial_port.is_open):
              logger.info("No active connection, attempting to connect...")
              if not self.connect():
                  error_msg = f"[MCP2Serial v{VERSION}] Failed to establish serial connection.\n"
                  error_msg += "Please check:\n"
                  error_msg += "1. Serial port is correctly configured in config.yaml\n"
                  error_msg += "2. Device is properly connected\n"
                  error_msg += "3. No other program is using the port"
                  return [types.TextContent(
                      type="text",
                      text=error_msg
                  )]
  
          # 准备命令
          cmd_str = command.command.format(**arguments)
          # 确保命令以\r\n结尾
          cmd_str = cmd_str.rstrip() + '\r\n'  # 移除可能的空白字符，强制添加\r\n
  
          cmd_bytes = cmd_str.encode()
          logger.info(f"Sending command: {cmd_str.strip()}")
          logger.info(f"Command bytes ({len(cmd_bytes)} bytes): {' '.join([f'0x{b:02X}' for b in cmd_bytes])}")
  
          if self.is_loopback:
              # 回环模式：直接返回发送的命令和OK响应
              responses = [
                  cmd_str.encode(),  # 命令回显
                  f"{config.response_start_string}\r\n".encode()  # OK响应
              ]
          else:
              # 清空缓冲区
              self.serial_port.reset_input_buffer()
              self.serial_port.reset_output_buffer()
  
              # 发送命令
              bytes_written = self.serial_port.write(cmd_bytes)
              logger.info(f"Wrote {bytes_written} bytes")
              self.serial_port.flush()
  
              # 等待一段时间确保命令被处理
              time.sleep(0.1)
  
              # 读取所有响应
              responses = []
              while self.serial_port.in_waiting:
                  response = self.serial_port.readline()
                  logger.info(f"Raw response: {response}")
                  if response:
                      responses.append(response)
  
          if not responses:
              logger.error("No response received within timeout")
              error_msg = f"[MCP2Serial v{VERSION}] Command timeout - no response within {self.read_timeout} second(s)\n"
              error_msg += f"Command sent: {cmd_str.strip()}\n"
              error_msg += f"Command bytes ({len(cmd_bytes)} bytes): {' '.join([f'0x{b:02X}' for b in cmd_bytes])}\n"
              error_msg += "Please check:\n"
              error_msg += "1. Device is powered and responding\n"
              error_msg += "2. Baud rate matches device settings\n"
              error_msg += "3. Serial connection is stable\n"
              return [types.TextContent(
                  type="text",
                  text=error_msg
              )]
  
          # 解码第一行响应
          first_response = responses[0]
          first_line = first_response.decode().strip()
          logger.info(f"Decoded first response: {first_line}")
  
          # 检查是否有第二行响应
          if len(responses) > 1:
              second_response = responses[1]
              if second_response.startswith(config.response_start_string.encode()):  # 使用配置的应答开始字符串
                  if command.need_parse:
                      return [types.TextContent(
                          type="text",
                          text=second_response.decode().strip()
                      )]
                  return []
  
          # 如果响应不是预期的格式，返回详细的错误信息
          error_msg = f"[MCP2Serial v{VERSION}] Command execution failed.\n"
          error_msg += f"Command sent: {cmd_str.strip()}\n"
          error_msg += f"Command bytes ({len(cmd_bytes)} bytes): {' '.join([f'0x{b:02X}' for b in cmd_bytes])}\n"
          error_msg += "Responses received:\n"
          for i, resp in enumerate(responses, 1):
              error_msg += f"{i}. Raw: {resp!r}\n   Decoded: {resp.decode().strip()}\n"
          error_msg += "\nPossible reasons:\n"
          error_msg += f"- Device echoed the command but did not send {config.response_start_string} response\n"
          error_msg += "- Command format may be incorrect\n"
          error_msg += "- Device may be in wrong mode\n"
          return [types.TextContent(
              type="text",
              text=error_msg
          )]
  
      except serial.SerialTimeoutException as e:
          logger.error(f"Serial timeout: {str(e)}")
          error_msg = f"[MCP2Serial v{VERSION}] Command timeout - {str(e)}\n"
          error_msg += "Please check:\n"
          error_msg += "1. Device is powered and responding\n"
          error_msg += "2. Baud rate matches device settings\n"
          error_msg += "3. Device is not busy with other operations"
          return [types.TextContent(
              type="text",
              text=error_msg
          )]
      except serial.SerialException as e:
          logger.error(f"Serial error: {str(e)}")
          error_msg = f"[MCP2Serial v{VERSION}] Serial communication failed - {str(e)}\n"
          error_msg += "Please check:\n"
          error_msg += "1. Serial port is correctly configured in config.yaml\n"
          error_msg += "2. Device is properly connected\n"
          error_msg += "3. No other program is using the port"
          return [types.TextContent(
              type="text",
              text=error_msg
          )]
  
  def close(self) -> None:
      """Close the serial port connection if open."""
      if self.serial_port and self.serial_port.is_open:
          try:
              self.serial_port.close()
              logger.info(f"Closed serial port connection: {self.serial_port.port}")
          except Exception as e:
              logger.error(f"Error closing port: {str(e)}")
          self.serial_port = None