open_port
Configure and open a serial port for device communication, supporting adjustable baud rate, data bits, parity, stop bits, flow control, timeouts, and automatic reconnection.
Instructions
打开指定串口，支持自定义配置参数
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| port | Yes | 串口路径，如 /dev/ttyUSB0 或 COM1 | |
| baudrate | No | 波特率，支持的值：[300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 38400, 57600, 115200, 230400, 460800, 921600] | |
| bytesize | No | 数据位，支持的值：[5, 6, 7, 8] | |
| parity | No | 校验位，支持的值：['N', 'E', 'O', 'M', 'S'] | N |
| stopbits | No | 停止位，支持的值：[1.0, 1.5, 2.0] | |
| flow_control | No | 流控制，支持的值：['none', 'hardware', 'software'] | none |
| read_timeout_ms | No | 读取超时（毫秒），范围 0-60000 | |
| write_timeout_ms | No | 写入超时（毫秒），范围 0-60000 | |
| auto_reconnect | No | 是否启用自动重连 |
Implementation Reference
- src/uart_mcp/tools/port_ops.py:25-64 (handler)The main handler function `open_port()` for the open_port tool. It takes serial port parameters (port, baudrate, bytesize, parity, stopbits, flow_control, timeouts, auto_reconnect), calls `SerialManager.open_port()`, and returns the port status as a dict.
def open_port( port: str, baudrate: int = DEFAULT_BAUDRATE, bytesize: int = DEFAULT_BYTESIZE, parity: str = DEFAULT_PARITY.value, stopbits: float = float(DEFAULT_STOPBITS.value), flow_control: str = DEFAULT_FLOW_CONTROL.value, read_timeout_ms: int = DEFAULT_TIMEOUT_MS, write_timeout_ms: int = DEFAULT_WRITE_TIMEOUT_MS, auto_reconnect: bool = True, ) -> dict[str, Any]: """打开串口 Args: port: 串口路径（如 /dev/ttyUSB0 或 COM1） baudrate: 波特率，默认 9600 bytesize: 数据位，默认 8 parity: 校验位，默认 N（无校验） stopbits: 停止位，默认 1 flow_control: 流控制，默认 none read_timeout_ms: 读取超时（毫秒），默认 1000 write_timeout_ms: 写入超时（毫秒），默认 1000 auto_reconnect: 是否启用自动重连，默认 True Returns: 串口状态信息 """ manager = get_serial_manager() status = manager.open_port( port=port, baudrate=baudrate, bytesize=bytesize, parity=parity, stopbits=stopbits, flow_control=flow_control, read_timeout_ms=read_timeout_ms, write_timeout_ms=write_timeout_ms, auto_reconnect=auto_reconnect, ) return status.to_dict() - The `OPEN_PORT_TOOL` dictionary defining the MCP tool schema (name, description, inputSchema) for the open_port tool. Specifies all parameters including port (required), baudrate, bytesize, parity, stopbits, flow_control, read/write timeouts, and auto_reconnect.
# 工具定义（用于 MCP 注册） OPEN_PORT_TOOL: dict[str, Any] = { "name": "open_port", "description": "打开指定串口，支持自定义配置参数", "inputSchema": { "type": "object", "properties": { "port": { "type": "string", "description": "串口路径，如 /dev/ttyUSB0 或 COM1", }, "baudrate": { "type": "integer", "description": f"波特率，支持的值：{list(SUPPORTED_BAUDRATES)}", "default": DEFAULT_BAUDRATE, }, "bytesize": { "type": "integer", "description": f"数据位，支持的值：{list(SUPPORTED_BYTESIZES)}", "default": DEFAULT_BYTESIZE, }, "parity": { "type": "string", "description": f"校验位，支持的值：{[p.value for p in Parity]}", "default": DEFAULT_PARITY.value, }, "stopbits": { "type": "number", "description": f"停止位，支持的值：{[s.value for s in StopBits]}", "default": float(DEFAULT_STOPBITS.value), }, "flow_control": { "type": "string", "description": f"流控制，支持的值：{[f.value for f in FlowControl]}", "default": DEFAULT_FLOW_CONTROL.value, }, "read_timeout_ms": { "type": "integer", "description": "读取超时（毫秒），范围 0-60000", "default": DEFAULT_TIMEOUT_MS, }, "write_timeout_ms": { "type": "integer", "description": "写入超时（毫秒），范围 0-60000", "default": DEFAULT_WRITE_TIMEOUT_MS, }, "auto_reconnect": { "type": "boolean", "description": "是否启用自动重连", "default": True, }, }, "required": ["port"], }, } - src/uart_mcp/server.py:72-76 (registration)Registration of the open_port tool in the `handle_list_tools()` function, exposing it as a `types.Tool` using the OPEN_PORT_TOOL schema.
types.Tool( name=OPEN_PORT_TOOL["name"], description=OPEN_PORT_TOOL["description"], inputSchema=OPEN_PORT_TOOL["inputSchema"], ), - src/uart_mcp/server.py:150-151 (registration)Route in `handle_call_tool()` that dispatches the 'open_port' name to the `open_port()` handler function with parsed arguments.
elif name == "open_port": result = open_port(**arguments) - The core implementation `SerialManager.open_port()` that validates parameters via `_validate_and_create_config()`, checks blacklist, creates a pyserial Serial object, and returns a PortStatus.
def open_port( self, port: str, baudrate: int | None = None, bytesize: int | None = None, parity: str | None = None, stopbits: float | None = None, flow_control: str | None = None, read_timeout_ms: int | None = None, write_timeout_ms: int | None = None, auto_reconnect: bool | None = None, ) -> PortStatus: """打开串口 Args: port: 串口路径 baudrate: 波特率（None 时使用配置默认值） bytesize: 数据位（None 时使用配置默认值） parity: 校验位（None 时使用配置默认值） stopbits: 停止位（None 时使用配置默认值） flow_control: 流控制（None 时使用配置默认值） read_timeout_ms: 读取超时（毫秒，None 时使用配置默认值） write_timeout_ms: 写入超时（毫秒，None 时使用配置默认值） auto_reconnect: 是否启用自动重连（None 时使用配置默认值） Returns: 串口状态 Raises: PortBlacklistedError: 串口在黑名单中 PortNotFoundError: 串口不存在 PortBusyError: 串口被占用 InvalidParamError: 参数无效 """ # 检查黑名单 blacklist = get_blacklist_manager() if blacklist.is_blacklisted(port): raise PortBlacklistedError(port) # 获取全局配置作为默认值 global_config = get_config_manager().config # 使用用户参数或配置默认值 final_baudrate = baudrate if baudrate is not None else global_config.baudrate final_bytesize = bytesize if bytesize is not None else global_config.bytesize final_parity = parity if parity is not None else global_config.parity final_stopbits = stopbits if stopbits is not None else global_config.stopbits # 流控：显式传入时优先使用传入值 if flow_control is None: final_flow_control = ( "software" if global_config.xonxoff else "hardware" if global_config.rtscts else "none" ) else: final_flow_control = flow_control cc = global_config rt = read_timeout_ms wt = write_timeout_ms ar = auto_reconnect final_read_timeout = cc.read_timeout if rt is None else rt final_write_timeout = cc.write_timeout if wt is None else wt final_auto_reconnect = cc.auto_reconnect if ar is None else ar # 验证参数 config = self._validate_and_create_config( final_baudrate, final_bytesize, final_parity, final_stopbits, final_flow_control, final_read_timeout, final_write_timeout, ) with self._lock: # 检查是否已打开（幂等操作） if port in self._ports: managed = self._ports[port] logger.info("串口已打开，返回当前状态：%s", port) return PortStatus( port=port, is_open=True, config=managed.config, connected=managed.is_connected, reconnecting=managed.reconnecting, ) # 打开串口 serial_obj = self._create_serial(port, config) managed = ManagedPort(port, serial_obj, config, final_auto_reconnect) self._ports[port] = managed logger.info("串口打开成功：%s", port) return PortStatus( port=port, is_open=True, config=config, connected=managed.is_connected, reconnecting=False, )