UART MCP Server

Instructions

Implementation Reference

• src/uart_mcp/tools/data_ops.py:13-40 (handler)

  Main handler function for the 'send_data' MCP tool. Handles input validation, data encoding (text or base64 binary), and delegates to SerialManager.send_data.

  def send_data(
      port: str,
      data: str,
      is_binary: bool = False,
  ) -> dict[str, Any]:
      """发送数据到串口
  
      Args:
          port: 串口路径
          data: 要发送的数据（文本模式为 UTF-8 字符串，二进制模式为 Base64 编码）
          is_binary: 是否为二进制模式
  
      Returns:
          发送结果，包含发送的字节数
      """
      manager = get_serial_manager()
  
      # 编码转换
      if is_binary:
          try:
              raw_data = base64.b64decode(data)
          except Exception as e:
              raise InvalidParamError("data", data, f"Base64 解码失败：{e}")
      else:
          raw_data = data.encode("utf-8")
  
      bytes_written = manager.send_data(port, raw_data)
      return {"success": True, "bytes_written": bytes_written}

• src/uart_mcp/tools/data_ops.py:73-95 (schema)

  JSON schema definition for the 'send_data' tool inputs, used for MCP tool registration.

  SEND_DATA_TOOL: dict[str, Any] = {
      "name": "send_data",
      "description": "向已打开的串口发送数据，支持文本和二进制模式",
      "inputSchema": {
          "type": "object",
          "properties": {
              "port": {
                  "type": "string",
                  "description": "串口路径，如 /dev/ttyUSB0 或 COM1",
              },
              "data": {
                  "type": "string",
                  "description": "要发送的数据（文本为UTF-8，二进制为Base64）",
              },
              "is_binary": {
                  "type": "boolean",
                  "description": "是否为二进制模式，True 时 data 为 Base64 编码",
                  "default": False,
              },
          },
          "required": ["port", "data"],
      },
  }

• src/uart_mcp/server.py:92-96 (registration)

  Registration of the 'send_data' tool in the MCP server's list_tools handler.

  types.Tool(
      name=SEND_DATA_TOOL["name"],
      description=SEND_DATA_TOOL["description"],
      inputSchema=SEND_DATA_TOOL["inputSchema"],
  ),

• src/uart_mcp/server.py:158-159 (handler)

  Dispatch logic in the MCP call_tool handler that invokes the send_data function.

  elif name == "send_data":
      result = send_data(**arguments)

• src/uart_mcp/serial_manager.py:619-646 (helper)

  Low-level helper method in SerialManager that performs the actual serial.write to send bytes to the port.

  def send_data(self, port: str, data: bytes) -> int:
      """发送原始字节数据
  
      Args:
          port: 串口路径
          data: 要发送的字节数据
  
      Returns:
          发送的字节数
  
      Raises:
          PortClosedError: 串口未打开
          WriteFailedError: 写入失败
      """
      with self._lock:
          if port not in self._ports:
              raise PortClosedError(port)
  
          managed = self._ports[port]
          try:
              result = managed.serial.write(data)
              bytes_written: int = result if result is not None else 0
              logger.debug("发送数据到串口 %s：%d 字节", port, bytes_written)
              return bytes_written
          except SerialException as e:
              logger.error("串口写入失败：%s - %s", port, e)
              raise WriteFailedError(port, str(e)) from e