qmp - QEMU Machine Protocol¶

QEMU Monitor Protocol (QMP) development library & tooling.

This package provides a fairly low-level class for communicating to QMP protocol servers, as implemented by QEMU, the QEMU Guest Agent, and the QEMU Storage Daemon. This library is not intended for production use.

QEMUMonitorProtocol is the primary class of interest, and all errors raised derive from QMPError.

class qemu.qmp.QEMUMonitorProtocol(address: Union[Tuple[str, int], str], server: bool = False, nickname: Optional[str] = None)[source]¶

Bases: object

Provide an API to connect to QEMU via QEMU Monitor Protocol (QMP) and then allow to handle commands and events.

T¶: alias of TypeVar(‘T’)

accept(timeout: Optional[float] = 15.0) → Dict[str, Any][source]¶

Await connection from QMP Monitor and perform capabilities negotiation.

@param timeout: timeout in seconds (nonnegative float number, or: None). The value passed will set the behavior of the underneath QMP socket as described in [1]. Default value is set to 15.0.

@return QMP greeting dict @raise OSError on socket connection errors @raise QMPConnectError if the greeting is not received @raise QMPCapabilitiesError if fails to negotiate capabilities

[1] https://docs.python.org/3/library/socket.html#socket.socket.settimeout

clear_events() → None [source]¶: Clear current list of pending events.

close() → None [source]¶: Close the socket and socket file.

cmd(name: str, args: Optional[Dict[str, object]] = None, cmd_id: Optional[object] = None) → Dict[str, Any][source]¶

Build a QMP command and send it to the QMP Monitor.

@param name: command name (string) @param args: command arguments (dict) @param cmd_id: command id (dict, list, string or int)

cmd_obj(qmp_cmd: Dict[str, Any]) → Dict[str, Any][source]¶

Send a QMP command to the QMP Monitor.

@param qmp_cmd: QMP command to be sent as a Python dict @return QMP response as a Python dict

command(cmd: str, **kwds: object) → object [source]¶: Build and send a QMP command to the monitor, report errors if any

connect(negotiate: bool = True) → Optional[Dict[str, Any]][source]¶

Connect to the QMP Monitor and perform capabilities negotiation.

@return QMP greeting dict, or None if negotiate is false @raise OSError on socket connection errors @raise QMPConnectError if the greeting is not received @raise QMPCapabilitiesError if fails to negotiate capabilities

get_events(wait: bool = False) → List[Dict[str, Any]][source]¶

Get a list of available QMP events.

@param wait (bool): block until an event is available. @param wait (float): If wait is a float, treat it as a timeout value.

@raise QMPTimeoutError: If a timeout float is provided and the timeout: period elapses.
@raise QMPConnectError: If wait is True but no events could be: retrieved or if some other error occurred.

@return The list of available QMP events.

get_sock_fd() → int [source]¶

Get the socket file descriptor.

@return The file descriptor number.

is_scm_available() → bool [source]¶

Check if the socket allows for SCM_RIGHTS.

@return True if SCM_RIGHTS is available, otherwise False.

logger = <Logger QMP (WARNING)>¶: Logger object for debugging messages

classmethod parse_address(address: str) → Union[Tuple[str, int], str][source]¶

Parse a string into a QMP address.

Figure out if the argument is in the port:host form. If it’s not, it’s probably a file path.

pull_event(wait: Union[bool, float] = False) → Optional[Dict[str, Any]][source]¶

Pulls a single event.

@param wait (bool): block until an event is available. @param wait (float): If wait is a float, treat it as a timeout value.

@raise QMPTimeoutError: If a timeout float is provided and the timeout: period elapses.
@raise QMPConnectError: If wait is True but no events could be: retrieved or if some other error occurred.

@return The first available QMP event, or None.

settimeout(timeout: Optional[float]) → None [source]¶