Skip to content

Client

erpc.client

HTTP client for eRPC runtime health and metrics endpoints.

Provides a lightweight client to query eRPC's health endpoint (on the server port) and Prometheus metrics endpoint (on the metrics port).

Examples:

Quick health check::

>>> client = ERPCClient("http://localhost:4000")
>>> client.is_healthy
True
>>> status = client.health()
>>> status.version
'0.0.49'

Fetch Prometheus metrics::

>>> metrics = client.metrics()
>>> metrics["erpc_requests_total{method=\"eth_call\"}"]
42.0

HealthStatus dataclass

Structured result from an eRPC health check.

Attributes:

Name Type Description
status str

Health status string (e.g. "ok").
uptime float

Process uptime in seconds.
version str

eRPC version string.
Source code in erpc/client.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
@dataclass(frozen=True)
class HealthStatus:
    """Structured result from an eRPC health check.

    Attributes:
        status: Health status string (e.g. ``"ok"``).
        uptime: Process uptime in seconds.
        version: eRPC version string.

    """

    status: str
    uptime: float
    version: str

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> HealthStatus:
        """Create a ``HealthStatus`` from a JSON-decoded dictionary.

        Missing fields are filled with sensible defaults so partial
        responses don't crash the client.

        Args:
            data: Decoded JSON body from the health endpoint.

        Returns:
            Populated ``HealthStatus`` instance.

        """
        return cls(
            status=str(data.get("status", "unknown")),
            uptime=float(data.get("uptime", 0.0)),
            version=str(data.get("version", "unknown")),
        )

from_dict classmethod 

from_dict(data: dict[str, Any]) -> HealthStatus

Create a HealthStatus from a JSON-decoded dictionary.

Missing fields are filled with sensible defaults so partial responses don't crash the client.

Parameters:

Name Type Description Default
data dict[str, Any]

Decoded JSON body from the health endpoint.

 required

Returns:

Type Description
HealthStatus

Populated HealthStatus instance.
Source code in erpc/client.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def from_dict(cls, data: dict[str, Any]) -> HealthStatus:
    """Create a ``HealthStatus`` from a JSON-decoded dictionary.

    Missing fields are filled with sensible defaults so partial
    responses don't crash the client.

    Args:
        data: Decoded JSON body from the health endpoint.

    Returns:
        Populated ``HealthStatus`` instance.

    """
    return cls(
        status=str(data.get("status", "unknown")),
        uptime=float(data.get("uptime", 0.0)),
        version=str(data.get("version", "unknown")),
    )

ERPCClient

Lightweight HTTP client for eRPC runtime endpoints.

Uses only :mod:urllib from the standard library — no extra dependencies.

Parameters:

Name Type Description Default
base_url str

Base URL of the eRPC server (e.g. http://localhost:4000).

 required
metrics_port int

Port for the Prometheus metrics endpoint. Defaults to 4001.

 4001
timeout int

HTTP request timeout in seconds. Defaults to 5.

 5

Examples:

>>> client = ERPCClient("http://localhost:4000")
>>> client.is_healthy
True
Source code in erpc/client.py 
 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
class ERPCClient:
    """Lightweight HTTP client for eRPC runtime endpoints.

    Uses only :mod:`urllib` from the standard library — no extra dependencies.

    Args:
        base_url: Base URL of the eRPC server (e.g. ``http://localhost:4000``).
        metrics_port: Port for the Prometheus metrics endpoint. Defaults to ``4001``.
        timeout: HTTP request timeout in seconds. Defaults to ``5``.

    Examples:
        >>> client = ERPCClient("http://localhost:4000")
        >>> client.is_healthy
        True

    """

    def __init__(
        self,
        base_url: str,
        metrics_port: int = 4001,
        timeout: int = 5,
    ) -> None:
        """Initialize the client.

        Args:
            base_url: Base URL of the eRPC server (e.g. ``http://localhost:4000``).
            metrics_port: Port for the Prometheus metrics endpoint.
            timeout: HTTP request timeout in seconds.

        """
        self.base_url: str = base_url.rstrip("/")
        self.metrics_port: int = metrics_port
        self.timeout: int = timeout

    def health(self) -> HealthStatus:
        """Fetch structured health status from eRPC.

        Queries the server's root endpoint and parses the JSON response
        into a :class:`HealthStatus` dataclass.

        Returns:
            Parsed health status.

        Raises:
            ERPCHealthCheckError: On connection failure, timeout, or unparseable response.

        """
        url = self.base_url + "/"
        try:
            with urlopen(url, timeout=self.timeout) as resp:
                body = resp.read()
        except (URLError, OSError, TimeoutError) as exc:
            raise ERPCHealthCheckError(f"Health check failed for {url}: {exc}") from exc

        try:
            data: dict[str, Any] = json.loads(body)
        except (json.JSONDecodeError, ValueError) as exc:
            raise ERPCHealthCheckError(f"Failed to parse health response: {exc}") from exc

        return HealthStatus.from_dict(data)

    @property
    def is_healthy(self) -> bool:
        """Quick boolean health check.

        Returns:
            ``True`` if the health endpoint responds successfully, ``False`` otherwise.

        """
        try:
            self.health()
        except ERPCHealthCheckError:
            return False
        return True

    def metrics(self) -> dict[str, float]:
        """Fetch and parse Prometheus metrics from eRPC.

        Queries the metrics endpoint and returns a flat dictionary of
        metric names (with labels) to their float values.

        Returns:
            Mapping of metric name (with labels) to value.

        Raises:
            ERPCHealthCheckError: On connection failure or timeout.

        """
        parsed = urlparse(self.base_url)
        host = parsed.hostname or "127.0.0.1"
        url = f"http://{host}:{self.metrics_port}/metrics"

        try:
            with urlopen(url, timeout=self.timeout) as resp:
                body = resp.read()
        except (URLError, OSError, TimeoutError) as exc:
            raise ERPCHealthCheckError(f"Failed to fetch metrics from {url}: {exc}") from exc

        return self._parse_prometheus(body.decode("utf-8", errors="replace"))

    @staticmethod
    def _parse_prometheus(text: str) -> dict[str, float]:
        """Parse Prometheus exposition format into a flat dict.

        Args:
            text: Raw Prometheus metrics text.

        Returns:
            Mapping of metric name (with labels) to float value.

        """
        result: dict[str, float] = {}
        for line in text.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            match = _METRIC_LINE_RE.match(line)
            if match:
                result[match.group(1)] = float(match.group(2))
        return result

is_healthy property 

is_healthy: bool

Quick boolean health check.

Returns:

Type Description
bool

True if the health endpoint responds successfully, False otherwise.

__init__ 

__init__(base_url: str, metrics_port: int = 4001, timeout: int = 5) -> None

Initialize the client.

Parameters:

Name Type Description Default
base_url str

Base URL of the eRPC server (e.g. http://localhost:4000).

 required
metrics_port int

Port for the Prometheus metrics endpoint.

 4001
timeout int

HTTP request timeout in seconds.

 5
Source code in erpc/client.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    base_url: str,
    metrics_port: int = 4001,
    timeout: int = 5,
) -> None:
    """Initialize the client.

    Args:
        base_url: Base URL of the eRPC server (e.g. ``http://localhost:4000``).
        metrics_port: Port for the Prometheus metrics endpoint.
        timeout: HTTP request timeout in seconds.

    """
    self.base_url: str = base_url.rstrip("/")
    self.metrics_port: int = metrics_port
    self.timeout: int = timeout

health 

health() -> HealthStatus

Fetch structured health status from eRPC.

Queries the server's root endpoint and parses the JSON response into a :class:HealthStatus dataclass.

Returns:

Type Description
HealthStatus

Parsed health status.

Raises:

Type Description
ERPCHealthCheckError

On connection failure, timeout, or unparseable response.
Source code in erpc/client.py 
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
def health(self) -> HealthStatus:
    """Fetch structured health status from eRPC.

    Queries the server's root endpoint and parses the JSON response
    into a :class:`HealthStatus` dataclass.

    Returns:
        Parsed health status.

    Raises:
        ERPCHealthCheckError: On connection failure, timeout, or unparseable response.

    """
    url = self.base_url + "/"
    try:
        with urlopen(url, timeout=self.timeout) as resp:
            body = resp.read()
    except (URLError, OSError, TimeoutError) as exc:
        raise ERPCHealthCheckError(f"Health check failed for {url}: {exc}") from exc

    try:
        data: dict[str, Any] = json.loads(body)
    except (json.JSONDecodeError, ValueError) as exc:
        raise ERPCHealthCheckError(f"Failed to parse health response: {exc}") from exc

    return HealthStatus.from_dict(data)

metrics 

metrics() -> dict[str, float]

Fetch and parse Prometheus metrics from eRPC.

Queries the metrics endpoint and returns a flat dictionary of metric names (with labels) to their float values.

Returns:

Type Description
dict[str, float]

Mapping of metric name (with labels) to value.

Raises:

Type Description
ERPCHealthCheckError

On connection failure or timeout.
Source code in erpc/client.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def metrics(self) -> dict[str, float]:
    """Fetch and parse Prometheus metrics from eRPC.

    Queries the metrics endpoint and returns a flat dictionary of
    metric names (with labels) to their float values.

    Returns:
        Mapping of metric name (with labels) to value.

    Raises:
        ERPCHealthCheckError: On connection failure or timeout.

    """
    parsed = urlparse(self.base_url)
    host = parsed.hostname or "127.0.0.1"
    url = f"http://{host}:{self.metrics_port}/metrics"

    try:
        with urlopen(url, timeout=self.timeout) as resp:
            body = resp.read()
    except (URLError, OSError, TimeoutError) as exc:
        raise ERPCHealthCheckError(f"Failed to fetch metrics from {url}: {exc}") from exc

    return self._parse_prometheus(body.decode("utf-8", errors="replace"))