Skip to content

XMRigAPI

A class to interact with the XMRig miner API.

Attributes:

Name Type Description
_miner_name str

Unique name for the miner.
_ip str

IP address of the XMRig API.
_port str

Port of the XMRig API.
_access_token Optional[str]

Access token for authorization.
_base_url str

Base URL for the XMRig API.
_json_rpc_url str

URL for the JSON RPC.
_summary_url str

URL for the summary endpoint.
_backends_url str

URL for the backends endpoint.
_config_url str

URL for the config endpoint.
_headers Dict[str, str]

Headers for all API/RPC requests.
_json_rpc_payload Dict[str, Union[str, int]]

Default payload to send with RPC request.
data XMRigProperties

Instance of XMRigProperties for accessing cached data.
Source code in xmrig/api.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class XMRigAPI:
    """
    A class to interact with the XMRig miner API.

    Attributes:
        _miner_name (str): Unique name for the miner.
        _ip (str): IP address of the XMRig API.
        _port (str): Port of the XMRig API.
        _access_token (Optional[str]): Access token for authorization.
        _base_url (str): Base URL for the XMRig API.
        _json_rpc_url (str): URL for the JSON RPC.
        _summary_url (str): URL for the summary endpoint.
        _backends_url (str): URL for the backends endpoint.
        _config_url (str): URL for the config endpoint.
        _headers (Dict[str, str]): Headers for all API/RPC requests.
        _json_rpc_payload (Dict[str, Union[str, int]]): Default payload to send with RPC request.
        data (XMRigProperties): Instance of XMRigProperties for accessing cached data.
    """

    def __init__(self, miner_name: str, ip: str, port: str, access_token: Optional[str] = None, tls_enabled: bool = False, db_url: Optional[str] = None) -> None:
        """
        Initializes the XMRig instance with the provided IP, port, and access token.

        The `ip` can be either an IP address or domain name with its TLD (e.g. `example.com`). The schema is not 
        required and the appropriate one will be chosen based on the `tls_enabled` value.

        Args:
            miner_name (str): A unique name for the miner.
            ip (str): IP address or domain of the XMRig API.
            port (str): Port of the XMRig API.
            access_token (Optional[str]): Access token for authorization. Defaults to None.
            tls_enabled (bool): TLS status of the miner/API. Defaults to False.
            db_url (Optional[str]): Database URL for storing miner data. Defaults to None.
        """
        self._miner_name = miner_name
        self._ip = ip
        self._port = port
        self._access_token = access_token
        self._tls_enabled = tls_enabled
        self._base_url = f"http://{ip}:{port}"
        if self._tls_enabled:
            self._base_url = f"https://{ip}:{port}"
        self._db_url = db_url
        self._json_rpc_url = f"{self._base_url}/json_rpc"
        self._summary_url = f"{self._base_url}/2/summary"
        self._backends_url = f"{self._base_url}/2/backends"
        self._config_url = f"{self._base_url}/2/config"
        self._headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Host": f"{self._base_url}",
            "Connection": "keep-alive",
            "Authorization": f"Bearer {self._access_token}"
        }
        self._json_rpc_payload = {
            "method": None,
            "jsonrpc": "2.0",
            "id": 1,
        }
        self.data = XMRigProperties(None, None, None, self._miner_name, self._db_url)
        self.get_all_responses()
        log.info(f"XMRigAPI initialized for {self._base_url}")

    def _update_properties_cache(self, response, endpoint) -> None:
        """
        Sets the properties for the XMRigAPI instance.
        """
        if endpoint == "summary":
            setattr(self.data, "_summary_response", response)
        if endpoint == "backends":
            setattr(self.data, "_backends_response", response)
        if endpoint == "config":
            setattr(self.data, "_config_response", response)

    def set_auth_header(self) -> bool:
        """
        Update the Authorization header for the HTTP requests.

        Returns:
            bool: True if the Authorization header was changed, or False if an error occurred.

        Raises:
            XMRigAuthorizationError: An error occurred setting the Authorization Header.
        """
        try:
            self._headers["Authorization"] = f"Bearer {self._access_token}"
            log.debug(f"Authorization header successfully changed.")
            return True
        except XMRigAuthorizationError as e:
            raise XMRigAuthorizationError(e, traceback.print_exc(), f"An error occurred setting the Authorization Header: {e}") from e

    def get_endpoint(self, endpoint: str) -> bool:
        """
        Updates the cached data from the specified XMRig API endpoint.

        Args:
            endpoint (str): The endpoint to fetch data from. Should be one of 'summary', 'backends', or 'config'.

        Returns:
            bool: True if the cached data is successfully updated or False if an error occurred.

        Raises:
            XMRigAuthorizationError: If an authorization error occurs.
            XMRigConnectionError: If a connection error occurs.
            XMRigAPIError: If a general API error occurs.
        """
        url_map = {
            "summary": self._summary_url,
            "backends": self._backends_url,
            "config": self._config_url
        }
        try:
            response = requests.get(url_map[endpoint], headers=self._headers)
            if response.status_code == 401:
                raise XMRigAuthorizationError(message = "401 UNAUTHORIZED")
            response.raise_for_status()
            try:
                json_response = response.json()
            except requests.exceptions.JSONDecodeError as e:
                json_response = None
                raise requests.exceptions.JSONDecodeError("JSON decode error", response.text, response.status_code)
            else:
                self._update_properties_cache(json_response, endpoint)
                log.debug(f"{endpoint.capitalize()} endpoint successfully fetched.")
                if self._db_url is not None:
                    if endpoint == "backends":
                        for backend in json_response:
                            prefix = ["cpu", "opencl", "cuda"][json_response.index(backend)]
                            XMRigDatabase.insert_data_to_db(backend, f"{self._miner_name}-{prefix}-backend", self._db_url)
                    else:
                        XMRigDatabase.insert_data_to_db(json_response, f"{self._miner_name}-{endpoint}", self._db_url)
                return True
        except requests.exceptions.JSONDecodeError as e:
            # INFO: Due to a bug in XMRig, the first 15 minutes a miner is running/restarted its backends 
            # INFO: endpoint will return a malformed JSON response, allow the program to continue running 
            # INFO: to bypass this bug for now until a fix is provided by the XMRig developers.
            log.error("Due to a bug in XMRig, the first 15 minutes a miner is running/restarted its backends endpoint will return a malformed JSON response. If that is the case then this error/warning can be safely ignored.")
            log.error(f"An error occurred decoding the {endpoint} response: {e}")
            return False
        except requests.exceptions.RequestException as e:
            raise XMRigConnectionError(e, traceback.print_exc(), f"An error occurred while connecting to {url_map[endpoint]}:") from e
        except XMRigAuthorizationError as e:
            raise XMRigAuthorizationError(e, traceback.print_exc(), f"An authorization error occurred updating the {endpoint}, please provide a valid access token:") from e
        except Exception as e:
            raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred updating the {endpoint}:") from e

    def post_config(self, config: Dict[str, Any]) -> bool:
        """
        Updates the miners config data via the XMRig API.

        Args:
            config (Dict[str, Any]): Configuration data to update.

        Returns:
            bool: True if the config was changed successfully, or False if an error occurred.

        Raises:
            XMRigAuthorizationError: If an authorization error occurs.
            XMRigConnectionError: If a connection error occurs.
            XMRigAPIError: If a general API error occurs.
        """
        try:
            response = requests.post(self._config_url, json=config, headers=self._headers)
            if response.status_code == 401:
                raise XMRigAuthorizationError()
            # Raise an HTTPError for bad responses (4xx and 5xx)
            response.raise_for_status()
            # Get the updated config data from the endpoint and update the cached data
            self.get_endpoint("config")
            log.debug(f"Config endpoint successfully updated.")
            return True
        except requests.exceptions.JSONDecodeError as e:
            raise requests.exceptions.JSONDecodeError("JSON decode error", response.text, response.status_code)
        except requests.exceptions.RequestException as e:
            raise XMRigConnectionError(e, traceback.print_exc(), f"An error occurred while connecting to {self._config_url}:") from e
        except XMRigAuthorizationError as e:
            raise XMRigAuthorizationError(e, traceback.print_exc(), f"An authorization error occurred posting the config, please provide a valid access token:") from e
        except Exception as e:
            raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred posting the config:") from e

    def get_all_responses(self) -> bool:
        """
        Retrieves all responses from the API.

        Returns:
            bool: True if successful, or False if an error occurred.

        Raises:
            XMRigAuthorizationError: If an authorization error occurs.
            XMRigConnectionError: If a connection error occurs.
            XMRigAPIError: If a general API error occurs.
        """
        summary_success = self.get_endpoint("summary")
        backends_success = self.get_endpoint("backends")
        config_success = self.get_endpoint("config")
        return summary_success and backends_success and config_success

    def perform_action(self, action: str) -> bool:
        """
        Controls the miner by performing the specified action.

        Args:
            action (str): The action to perform. Valid actions are 'pause', 'resume', 'stop', 'start'.

        Returns:
            bool: True if the action was successfully performed, or False if an error occurred.

        Raises:
            XMRigConnectionError: If a connection error occurs.
            XMRigAPIError: If a general API error occurs.
        """
        try:
            # TODO: The `start` json RPC method is not implemented by XMRig yet, use alternative implementation 
            # TODO: until PR 3030 is merged, see the following issues and PRs for more information: 
            # TODO: https://github.com/xmrig/xmrig/issues/2826#issuecomment-1146465641
            # TODO: https://github.com/xmrig/xmrig/issues/3220#issuecomment-1450691309
            # TODO: https://github.com/xmrig/xmrig/pull/3030
            if action == "start":
                self.get_endpoint("config")
                self.post_config(self.data._config_response)
                log.debug(f"Miner successfully started.")
            else:
                url = f"{self._json_rpc_url}"
                payload = self._json_rpc_payload
                payload["method"] = action
                response = requests.post(url, json=payload, headers=self._headers)
                response.raise_for_status()
                log.debug(f"Miner successfully {action}ed.")
            return True
        except requests.exceptions.RequestException as e:
            raise XMRigConnectionError(e, traceback.print_exc(), f"A connection error occurred {action}ing the miner:") from e
        except Exception as e:
            raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred {action}ing the miner:") from e

__init__(miner_name, ip, port, access_token=None, tls_enabled=False, db_url=None)

Initializes the XMRig instance with the provided IP, port, and access token.

The ip can be either an IP address or domain name with its TLD (e.g. example.com). The schema is not required and the appropriate one will be chosen based on the tls_enabled value.

Parameters:

Name Type Description Default
miner_name str

A unique name for the miner.

 required
ip str

IP address or domain of the XMRig API.

 required
port str

Port of the XMRig API.

 required
access_token Optional[str]

Access token for authorization. Defaults to None.

 None
tls_enabled bool

TLS status of the miner/API. Defaults to False.

 False
db_url Optional[str]

Database URL for storing miner data. Defaults to None.

 None
Source code in xmrig/api.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(self, miner_name: str, ip: str, port: str, access_token: Optional[str] = None, tls_enabled: bool = False, db_url: Optional[str] = None) -> None:
    """
    Initializes the XMRig instance with the provided IP, port, and access token.

    The `ip` can be either an IP address or domain name with its TLD (e.g. `example.com`). The schema is not 
    required and the appropriate one will be chosen based on the `tls_enabled` value.

    Args:
        miner_name (str): A unique name for the miner.
        ip (str): IP address or domain of the XMRig API.
        port (str): Port of the XMRig API.
        access_token (Optional[str]): Access token for authorization. Defaults to None.
        tls_enabled (bool): TLS status of the miner/API. Defaults to False.
        db_url (Optional[str]): Database URL for storing miner data. Defaults to None.
    """
    self._miner_name = miner_name
    self._ip = ip
    self._port = port
    self._access_token = access_token
    self._tls_enabled = tls_enabled
    self._base_url = f"http://{ip}:{port}"
    if self._tls_enabled:
        self._base_url = f"https://{ip}:{port}"
    self._db_url = db_url
    self._json_rpc_url = f"{self._base_url}/json_rpc"
    self._summary_url = f"{self._base_url}/2/summary"
    self._backends_url = f"{self._base_url}/2/backends"
    self._config_url = f"{self._base_url}/2/config"
    self._headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Host": f"{self._base_url}",
        "Connection": "keep-alive",
        "Authorization": f"Bearer {self._access_token}"
    }
    self._json_rpc_payload = {
        "method": None,
        "jsonrpc": "2.0",
        "id": 1,
    }
    self.data = XMRigProperties(None, None, None, self._miner_name, self._db_url)
    self.get_all_responses()
    log.info(f"XMRigAPI initialized for {self._base_url}")

get_all_responses()

Retrieves all responses from the API.

Returns:

Name Type Description
bool bool

True if successful, or False if an error occurred.

Raises:

Type Description
XMRigAuthorizationError

If an authorization error occurs.
XMRigConnectionError

If a connection error occurs.
XMRigAPIError

If a general API error occurs.
Source code in xmrig/api.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def get_all_responses(self) -> bool:
    """
    Retrieves all responses from the API.

    Returns:
        bool: True if successful, or False if an error occurred.

    Raises:
        XMRigAuthorizationError: If an authorization error occurs.
        XMRigConnectionError: If a connection error occurs.
        XMRigAPIError: If a general API error occurs.
    """
    summary_success = self.get_endpoint("summary")
    backends_success = self.get_endpoint("backends")
    config_success = self.get_endpoint("config")
    return summary_success and backends_success and config_success

get_endpoint(endpoint)

Updates the cached data from the specified XMRig API endpoint.

Parameters:

Name Type Description Default
endpoint str

The endpoint to fetch data from. Should be one of 'summary', 'backends', or 'config'.

 required

Returns:

Name Type Description
bool bool

True if the cached data is successfully updated or False if an error occurred.

Raises:

Type Description
XMRigAuthorizationError

If an authorization error occurs.
XMRigConnectionError

If a connection error occurs.
XMRigAPIError

If a general API error occurs.
Source code in xmrig/api.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def get_endpoint(self, endpoint: str) -> bool:
    """
    Updates the cached data from the specified XMRig API endpoint.

    Args:
        endpoint (str): The endpoint to fetch data from. Should be one of 'summary', 'backends', or 'config'.

    Returns:
        bool: True if the cached data is successfully updated or False if an error occurred.

    Raises:
        XMRigAuthorizationError: If an authorization error occurs.
        XMRigConnectionError: If a connection error occurs.
        XMRigAPIError: If a general API error occurs.
    """
    url_map = {
        "summary": self._summary_url,
        "backends": self._backends_url,
        "config": self._config_url
    }
    try:
        response = requests.get(url_map[endpoint], headers=self._headers)
        if response.status_code == 401:
            raise XMRigAuthorizationError(message = "401 UNAUTHORIZED")
        response.raise_for_status()
        try:
            json_response = response.json()
        except requests.exceptions.JSONDecodeError as e:
            json_response = None
            raise requests.exceptions.JSONDecodeError("JSON decode error", response.text, response.status_code)
        else:
            self._update_properties_cache(json_response, endpoint)
            log.debug(f"{endpoint.capitalize()} endpoint successfully fetched.")
            if self._db_url is not None:
                if endpoint == "backends":
                    for backend in json_response:
                        prefix = ["cpu", "opencl", "cuda"][json_response.index(backend)]
                        XMRigDatabase.insert_data_to_db(backend, f"{self._miner_name}-{prefix}-backend", self._db_url)
                else:
                    XMRigDatabase.insert_data_to_db(json_response, f"{self._miner_name}-{endpoint}", self._db_url)
            return True
    except requests.exceptions.JSONDecodeError as e:
        # INFO: Due to a bug in XMRig, the first 15 minutes a miner is running/restarted its backends 
        # INFO: endpoint will return a malformed JSON response, allow the program to continue running 
        # INFO: to bypass this bug for now until a fix is provided by the XMRig developers.
        log.error("Due to a bug in XMRig, the first 15 minutes a miner is running/restarted its backends endpoint will return a malformed JSON response. If that is the case then this error/warning can be safely ignored.")
        log.error(f"An error occurred decoding the {endpoint} response: {e}")
        return False
    except requests.exceptions.RequestException as e:
        raise XMRigConnectionError(e, traceback.print_exc(), f"An error occurred while connecting to {url_map[endpoint]}:") from e
    except XMRigAuthorizationError as e:
        raise XMRigAuthorizationError(e, traceback.print_exc(), f"An authorization error occurred updating the {endpoint}, please provide a valid access token:") from e
    except Exception as e:
        raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred updating the {endpoint}:") from e

perform_action(action)

Controls the miner by performing the specified action.

Parameters:

Name Type Description Default
action str

The action to perform. Valid actions are 'pause', 'resume', 'stop', 'start'.

 required

Returns:

Name Type Description
bool bool

True if the action was successfully performed, or False if an error occurred.

Raises:

Type Description
XMRigConnectionError

If a connection error occurs.
XMRigAPIError

If a general API error occurs.
Source code in xmrig/api.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def perform_action(self, action: str) -> bool:
    """
    Controls the miner by performing the specified action.

    Args:
        action (str): The action to perform. Valid actions are 'pause', 'resume', 'stop', 'start'.

    Returns:
        bool: True if the action was successfully performed, or False if an error occurred.

    Raises:
        XMRigConnectionError: If a connection error occurs.
        XMRigAPIError: If a general API error occurs.
    """
    try:
        # TODO: The `start` json RPC method is not implemented by XMRig yet, use alternative implementation 
        # TODO: until PR 3030 is merged, see the following issues and PRs for more information: 
        # TODO: https://github.com/xmrig/xmrig/issues/2826#issuecomment-1146465641
        # TODO: https://github.com/xmrig/xmrig/issues/3220#issuecomment-1450691309
        # TODO: https://github.com/xmrig/xmrig/pull/3030
        if action == "start":
            self.get_endpoint("config")
            self.post_config(self.data._config_response)
            log.debug(f"Miner successfully started.")
        else:
            url = f"{self._json_rpc_url}"
            payload = self._json_rpc_payload
            payload["method"] = action
            response = requests.post(url, json=payload, headers=self._headers)
            response.raise_for_status()
            log.debug(f"Miner successfully {action}ed.")
        return True
    except requests.exceptions.RequestException as e:
        raise XMRigConnectionError(e, traceback.print_exc(), f"A connection error occurred {action}ing the miner:") from e
    except Exception as e:
        raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred {action}ing the miner:") from e

post_config(config)

Updates the miners config data via the XMRig API.

Parameters:

Name Type Description Default
config Dict[str, Any]

Configuration data to update.

 required

Returns:

Name Type Description
bool bool

True if the config was changed successfully, or False if an error occurred.

Raises:

Type Description
XMRigAuthorizationError

If an authorization error occurs.
XMRigConnectionError

If a connection error occurs.
XMRigAPIError

If a general API error occurs.
Source code in xmrig/api.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def post_config(self, config: Dict[str, Any]) -> bool:
    """
    Updates the miners config data via the XMRig API.

    Args:
        config (Dict[str, Any]): Configuration data to update.

    Returns:
        bool: True if the config was changed successfully, or False if an error occurred.

    Raises:
        XMRigAuthorizationError: If an authorization error occurs.
        XMRigConnectionError: If a connection error occurs.
        XMRigAPIError: If a general API error occurs.
    """
    try:
        response = requests.post(self._config_url, json=config, headers=self._headers)
        if response.status_code == 401:
            raise XMRigAuthorizationError()
        # Raise an HTTPError for bad responses (4xx and 5xx)
        response.raise_for_status()
        # Get the updated config data from the endpoint and update the cached data
        self.get_endpoint("config")
        log.debug(f"Config endpoint successfully updated.")
        return True
    except requests.exceptions.JSONDecodeError as e:
        raise requests.exceptions.JSONDecodeError("JSON decode error", response.text, response.status_code)
    except requests.exceptions.RequestException as e:
        raise XMRigConnectionError(e, traceback.print_exc(), f"An error occurred while connecting to {self._config_url}:") from e
    except XMRigAuthorizationError as e:
        raise XMRigAuthorizationError(e, traceback.print_exc(), f"An authorization error occurred posting the config, please provide a valid access token:") from e
    except Exception as e:
        raise XMRigAPIError(e, traceback.print_exc(), f"An error occurred posting the config:") from e

set_auth_header()

Update the Authorization header for the HTTP requests.

Returns:

Name Type Description
bool bool

True if the Authorization header was changed, or False if an error occurred.

Raises:

Type Description
XMRigAuthorizationError

An error occurred setting the Authorization Header.
Source code in xmrig/api.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def set_auth_header(self) -> bool:
    """
    Update the Authorization header for the HTTP requests.

    Returns:
        bool: True if the Authorization header was changed, or False if an error occurred.

    Raises:
        XMRigAuthorizationError: An error occurred setting the Authorization Header.
    """
    try:
        self._headers["Authorization"] = f"Bearer {self._access_token}"
        log.debug(f"Authorization header successfully changed.")
        return True
    except XMRigAuthorizationError as e:
        raise XMRigAuthorizationError(e, traceback.print_exc(), f"An error occurred setting the Authorization Header: {e}") from e