Skip to content

Config

erpc.config

eRPC configuration generation and loading.

CacheConfig dataclass

Memory cache configuration for eRPC.

Attributes:

Name Type Description
max_items int

Maximum number of items to store in memory cache.
method_ttls dict[str, int]

Per-method TTL overrides in seconds. Use 0 to disable caching.
Source code in erpc/config.py 
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass
class CacheConfig:
    """Memory cache configuration for eRPC.

    Attributes:
        max_items: Maximum number of items to store in memory cache.
        method_ttls: Per-method TTL overrides in seconds. Use ``0`` to disable caching.

    """

    max_items: int = 10_000
    method_ttls: dict[str, int] = field(default_factory=dict)

ERPCConfig dataclass

eRPC configuration builder.

Generates an erpc.yaml config file from Python dataclasses.

Parameters:

Name Type Description Default
project_id str

Unique project identifier for eRPC routing.

 'py-erpc'
upstreams dict[int, list[str]]

Mapping of chain ID to list of RPC endpoint URLs.

 dict()
server_host str

eRPC server bind address (legacy, use server instead).

 '127.0.0.1'
server_port int

eRPC server listen port (legacy, use server instead).

 4000
metrics_host str

Metrics endpoint bind address (legacy, use metrics instead).

 '127.0.0.1'
metrics_port int

Metrics endpoint listen port (legacy, use metrics instead).

 4001
log_level str

Logging verbosity (trace, debug, info, warn, error).

 'warn'
cache CacheConfig

Memory cache configuration.

 CacheConfig()
server ServerConfig | None

Full server configuration. Takes precedence over server_host/server_port.

 None
metrics MetricsConfig | None

Full metrics configuration. Takes precedence over metrics_host/metrics_port.

 None
upstream_defaults UpstreamConfig | None

Default configuration applied to all upstreams.

 None
rich_upstreams list[UpstreamConfig]

List of fully-configured UpstreamConfig objects.

 list()

Examples:

>>> config = ERPCConfig(upstreams={1: ["https://eth.llamarpc.com"]})
>>> config.endpoint_url(1)
'http://127.0.0.1:4000/py-erpc/evm/1'
Source code in erpc/config.py 
 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
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
@dataclass
class ERPCConfig:
    """eRPC configuration builder.

    Generates an ``erpc.yaml`` config file from Python dataclasses.

    Args:
        project_id: Unique project identifier for eRPC routing.
        upstreams: Mapping of chain ID to list of RPC endpoint URLs.
        server_host: eRPC server bind address (legacy, use ``server`` instead).
        server_port: eRPC server listen port (legacy, use ``server`` instead).
        metrics_host: Metrics endpoint bind address (legacy, use ``metrics`` instead).
        metrics_port: Metrics endpoint listen port (legacy, use ``metrics`` instead).
        log_level: Logging verbosity (trace, debug, info, warn, error).
        cache: Memory cache configuration.
        server: Full server configuration. Takes precedence over ``server_host``/``server_port``.
        metrics: Full metrics configuration. Takes precedence over
            ``metrics_host``/``metrics_port``.
        upstream_defaults: Default configuration applied to all upstreams.
        rich_upstreams: List of fully-configured UpstreamConfig objects.

    Examples:
        >>> config = ERPCConfig(upstreams={1: ["https://eth.llamarpc.com"]})
        >>> config.endpoint_url(1)
        'http://127.0.0.1:4000/py-erpc/evm/1'

    """

    project_id: str = "py-erpc"
    upstreams: dict[int, list[str]] = field(default_factory=dict)
    server_host: str = "127.0.0.1"
    server_port: int = 4000
    metrics_host: str = "127.0.0.1"
    metrics_port: int = 4001
    log_level: str = "warn"
    cache: CacheConfig = field(default_factory=CacheConfig)
    providers: list[Provider] = field(default_factory=list)
    server: ServerConfig | None = None
    metrics: MetricsConfig | None = None
    auth: AuthConfig | None = None
    database: DatabaseConfig | None = None
    networks: list[NetworkConfig] = field(default_factory=list)
    network_defaults: NetworkConfig | None = None
    upstream_defaults: UpstreamConfig | None = None
    rich_upstreams: list[UpstreamConfig] = field(default_factory=list)

    @classmethod
    def from_yaml(cls, path: str | Path) -> ERPCConfig:
        """Load an ERPCConfig from an existing erpc.yaml file.

        Args:
            path: Path to the YAML configuration file.

        Returns:
            Parsed configuration object.

        Raises:
            FileNotFoundError: If the file does not exist.
            ERPCConfigError: If the YAML is malformed or missing required keys.

        Examples:
            >>> config = ERPCConfig.from_yaml("erpc.yaml")
            >>> config.project_id
            'my-project'

        """
        path = Path(path)
        if not path.exists():
            raise FileNotFoundError(f"Config file not found: {path}")
        try:
            data = yaml.safe_load(path.read_text())
        except yaml.YAMLError as exc:
            raise ERPCConfigError(f"Invalid YAML in {path}: {exc}") from exc
        if not isinstance(data, dict):
            raise ERPCConfigError(f"Expected YAML mapping, got {type(data).__name__}")
        return cls.from_dict(data)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ERPCConfig:
        """Construct an ERPCConfig from a plain dictionary.

        Args:
            data: Dictionary matching the eRPC YAML schema.

        Returns:
            Parsed configuration object.

        Raises:
            ERPCConfigError: If the dictionary is missing required keys.

        Examples:
            >>> config = ERPCConfig.from_dict({"projects": [{"id": "x", "networks": []}]})
            >>> config.project_id
            'x'

        """
        if "projects" not in data or not data["projects"]:
            raise ERPCConfigError("Config must contain a non-empty 'projects' list")

        project = data["projects"][0]
        server = data.get("server", {})
        metrics = data.get("metrics", {})

        # Parse upstreams from project-level upstreams list
        upstreams: dict[int, list[str]] = {}
        for upstream in project.get("upstreams", []):
            evm = upstream.get("evm", {})
            chain_id = evm.get("chainId")
            if chain_id is not None:
                upstreams.setdefault(chain_id, []).append(upstream["endpoint"])

        # Fallback: parse upstreams from networks (legacy format)
        if not upstreams:
            for network in project.get("networks", []):
                evm = network.get("evm", {})
                chain_id = evm.get("chainId")
                if chain_id is not None:
                    endpoints = [u["endpoint"] for u in network.get("upstreams", [])]
                    upstreams[chain_id] = endpoints

        # Parse cache config
        cache_config = CacheConfig()
        cache_data = project.get("cacheConfig", {})
        connectors = cache_data.get("connectors", [])
        if connectors:
            memory = connectors[0].get("memory", {})
            max_items = memory.get("maxItems", 10_000)
            cache_config = CacheConfig(max_items=max_items)

        return cls(
            project_id=project.get("id", "py-erpc"),
            upstreams=upstreams,
            server_host=server.get("httpHostV4", server.get("httpHost", "127.0.0.1")),
            server_port=server.get("httpPort", 4000),
            metrics_host=metrics.get("hostV4", metrics.get("host", "127.0.0.1")),
            metrics_port=metrics.get("port", 4001),
            log_level=data.get("logLevel", "warn"),
            cache=cache_config,
        )

    def validate(self) -> None:
        """Validate this configuration, raising on errors and warning on issues.

        Raises:
            ERPCConfigError: If the configuration is invalid.

        Examples:
            >>> config = ERPCConfig(upstreams={1: ["https://rpc.example.com"]})
            >>> config.validate()  # no error

        """
        if not self.upstreams:
            warnings.warn(
                "No upstreams configured — eRPC will have no RPC endpoints to proxy",
                UserWarning,
                stacklevel=2,
            )

    @property
    def _effective_host(self) -> str:
        """Resolve the effective server host."""
        if self.server is not None:
            return self.server.http_host
        return self.server_host

    @property
    def _effective_port(self) -> int:
        """Resolve the effective server port."""
        if self.server is not None:
            return self.server.http_port
        return self.server_port

    @property
    def health_url(self) -> str:
        """HTTP URL for the eRPC health endpoint."""
        return f"http://{self._effective_host}:{self._effective_port}/"

    def endpoint_url(self, chain_id: int) -> str:
        """Get the proxied endpoint URL for a specific chain.

        Args:
            chain_id: EVM chain identifier.

        Returns:
            Full URL for the proxied RPC endpoint.

        """
        return (
            f"http://{self._effective_host}:{self._effective_port}"
            f"/{self.project_id}/evm/{chain_id}"
        )

    def to_yaml(self) -> str:
        """Generate eRPC YAML configuration string.

        Returns:
            YAML-formatted configuration document.

        """
        if self.server is not None:
            server_dict = self.server.to_dict()
        else:
            server_dict = {
                "httpHostV4": self.server_host,
                "httpPort": self.server_port,
                "maxTimeout": "60s",
            }

        if self.metrics is not None:
            metrics_dict = self.metrics.to_dict()
        else:
            metrics_dict = {
                "enabled": True,
                "hostV4": self.metrics_host,
                "port": self.metrics_port,
            }

        doc: dict[str, Any] = {
            "logLevel": self.log_level,
            "server": server_dict,
            "metrics": metrics_dict,
            "projects": [self._build_project()],
        }
        return yaml.dump(doc, default_flow_style=False, sort_keys=False)

    def write(self, path: Path | None = None) -> Path:
        """Write config to a YAML file.

        Args:
            path: Destination file path. If ``None``, writes to a temporary file.

        Returns:
            Path to the written configuration file.

        """
        if path is None:
            fd = tempfile.NamedTemporaryFile(
                mode="w", suffix=".yaml", prefix="erpc-", delete=False
            )
            fd.write(self.to_yaml())
            fd.close()
            return Path(fd.name)
        else:
            path = Path(path)
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(self.to_yaml())
            return path

    def _build_project(self) -> dict[str, Any]:
        """Build a single eRPC project definition."""
        # Index explicit NetworkConfig objects by chain_id
        net_configs: dict[int, NetworkConfig] = {n.chain_id: n for n in self.networks}

        networks: list[dict[str, Any]] = []
        all_upstreams: list[dict[str, Any]] = []

        for chain_id, endpoints in self.upstreams.items():
            for i, url in enumerate(endpoints):
                all_upstreams.append(
                    {
                        "endpoint": url,
                        "id": f"upstream-{chain_id}-{i}",
                        "evm": {"chainId": chain_id},
                    }
                )

            # Use explicit NetworkConfig if provided, otherwise minimal
            if chain_id in net_configs:
                network: dict[str, Any] = net_configs[chain_id].to_dict()
            else:
                network = {
                    "architecture": "evm",
                    "evm": {"chainId": chain_id},
                }

            if self.cache.method_ttls:
                network["policies"] = self._build_cache_policies()

            networks.append(network)

        project: dict[str, Any] = {
            "id": self.project_id,
            "networks": networks,
            "upstreams": all_upstreams,
        }

        if self.upstream_defaults is not None:
            project["upstreamDefaults"] = self.upstream_defaults.to_dict()

        if self.rich_upstreams:
            project["upstreams"].extend(ru.to_dict() for ru in self.rich_upstreams)

        if self.network_defaults is not None:
            project["networkDefaults"] = self.network_defaults.to_defaults_dict()

        if self.providers:
            project["providers"] = [p.to_dict() for p in self.providers]

        if self.database is not None:
            project["database"] = self.database.to_dict()

        if self.auth is not None:
            project["auth"] = self.auth.to_dict()

        return project

    def _build_cache_policies(self) -> list[dict[str, Any]]:
        """Build cache policy rules from method TTL overrides."""
        policies: list[dict[str, Any]] = []
        for method, ttl in self.cache.method_ttls.items():
            if ttl == 0:
                policies.append(
                    {
                        "method": method,
                        "finality": "unfinalized",
                        "cache": {"empty": "skip", "ttl": "0s"},
                    }
                )
            else:
                policies.append(
                    {
                        "method": method,
                        "finality": "unfinalized",
                        "cache": {"ttl": f"{ttl}s"},
                    }
                )
        return policies

health_url property 

health_url: str

HTTP URL for the eRPC health endpoint.

from_yaml classmethod 

from_yaml(path: str | Path) -> ERPCConfig

Load an ERPCConfig from an existing erpc.yaml file.

Parameters:

Name Type Description Default
path str | Path

Path to the YAML configuration file.

 required

Returns:

Type Description
ERPCConfig

Parsed configuration object.

Raises:

Type Description
FileNotFoundError

If the file does not exist.
ERPCConfigError

If the YAML is malformed or missing required keys.

Examples:

>>> config = ERPCConfig.from_yaml("erpc.yaml")
>>> config.project_id
'my-project'
Source code in erpc/config.py 
 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
@classmethod
def from_yaml(cls, path: str | Path) -> ERPCConfig:
    """Load an ERPCConfig from an existing erpc.yaml file.

    Args:
        path: Path to the YAML configuration file.

    Returns:
        Parsed configuration object.

    Raises:
        FileNotFoundError: If the file does not exist.
        ERPCConfigError: If the YAML is malformed or missing required keys.

    Examples:
        >>> config = ERPCConfig.from_yaml("erpc.yaml")
        >>> config.project_id
        'my-project'

    """
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {path}")
    try:
        data = yaml.safe_load(path.read_text())
    except yaml.YAMLError as exc:
        raise ERPCConfigError(f"Invalid YAML in {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise ERPCConfigError(f"Expected YAML mapping, got {type(data).__name__}")
    return cls.from_dict(data)

from_dict classmethod 

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

Construct an ERPCConfig from a plain dictionary.

Parameters:

Name Type Description Default
data dict[str, Any]

Dictionary matching the eRPC YAML schema.

 required

Returns:

Type Description
ERPCConfig

Parsed configuration object.

Raises:

Type Description
ERPCConfigError

If the dictionary is missing required keys.

Examples:

>>> config = ERPCConfig.from_dict({"projects": [{"id": "x", "networks": []}]})
>>> config.project_id
'x'
Source code in erpc/config.py 
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
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ERPCConfig:
    """Construct an ERPCConfig from a plain dictionary.

    Args:
        data: Dictionary matching the eRPC YAML schema.

    Returns:
        Parsed configuration object.

    Raises:
        ERPCConfigError: If the dictionary is missing required keys.

    Examples:
        >>> config = ERPCConfig.from_dict({"projects": [{"id": "x", "networks": []}]})
        >>> config.project_id
        'x'

    """
    if "projects" not in data or not data["projects"]:
        raise ERPCConfigError("Config must contain a non-empty 'projects' list")

    project = data["projects"][0]
    server = data.get("server", {})
    metrics = data.get("metrics", {})

    # Parse upstreams from project-level upstreams list
    upstreams: dict[int, list[str]] = {}
    for upstream in project.get("upstreams", []):
        evm = upstream.get("evm", {})
        chain_id = evm.get("chainId")
        if chain_id is not None:
            upstreams.setdefault(chain_id, []).append(upstream["endpoint"])

    # Fallback: parse upstreams from networks (legacy format)
    if not upstreams:
        for network in project.get("networks", []):
            evm = network.get("evm", {})
            chain_id = evm.get("chainId")
            if chain_id is not None:
                endpoints = [u["endpoint"] for u in network.get("upstreams", [])]
                upstreams[chain_id] = endpoints

    # Parse cache config
    cache_config = CacheConfig()
    cache_data = project.get("cacheConfig", {})
    connectors = cache_data.get("connectors", [])
    if connectors:
        memory = connectors[0].get("memory", {})
        max_items = memory.get("maxItems", 10_000)
        cache_config = CacheConfig(max_items=max_items)

    return cls(
        project_id=project.get("id", "py-erpc"),
        upstreams=upstreams,
        server_host=server.get("httpHostV4", server.get("httpHost", "127.0.0.1")),
        server_port=server.get("httpPort", 4000),
        metrics_host=metrics.get("hostV4", metrics.get("host", "127.0.0.1")),
        metrics_port=metrics.get("port", 4001),
        log_level=data.get("logLevel", "warn"),
        cache=cache_config,
    )

validate 

validate() -> None

Validate this configuration, raising on errors and warning on issues.

Raises:

Type Description
ERPCConfigError

If the configuration is invalid.

Examples:

>>> config = ERPCConfig(upstreams={1: ["https://rpc.example.com"]})
>>> config.validate()  # no error
Source code in erpc/config.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def validate(self) -> None:
    """Validate this configuration, raising on errors and warning on issues.

    Raises:
        ERPCConfigError: If the configuration is invalid.

    Examples:
        >>> config = ERPCConfig(upstreams={1: ["https://rpc.example.com"]})
        >>> config.validate()  # no error

    """
    if not self.upstreams:
        warnings.warn(
            "No upstreams configured — eRPC will have no RPC endpoints to proxy",
            UserWarning,
            stacklevel=2,
        )

endpoint_url 

endpoint_url(chain_id: int) -> str

Get the proxied endpoint URL for a specific chain.

Parameters:

Name Type Description Default
chain_id int

EVM chain identifier.

 required

Returns:

Type Description
str

Full URL for the proxied RPC endpoint.
Source code in erpc/config.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def endpoint_url(self, chain_id: int) -> str:
    """Get the proxied endpoint URL for a specific chain.

    Args:
        chain_id: EVM chain identifier.

    Returns:
        Full URL for the proxied RPC endpoint.

    """
    return (
        f"http://{self._effective_host}:{self._effective_port}"
        f"/{self.project_id}/evm/{chain_id}"
    )

to_yaml 

to_yaml() -> str

Generate eRPC YAML configuration string.

Returns:

Type Description
str

YAML-formatted configuration document.
Source code in erpc/config.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def to_yaml(self) -> str:
    """Generate eRPC YAML configuration string.

    Returns:
        YAML-formatted configuration document.

    """
    if self.server is not None:
        server_dict = self.server.to_dict()
    else:
        server_dict = {
            "httpHostV4": self.server_host,
            "httpPort": self.server_port,
            "maxTimeout": "60s",
        }

    if self.metrics is not None:
        metrics_dict = self.metrics.to_dict()
    else:
        metrics_dict = {
            "enabled": True,
            "hostV4": self.metrics_host,
            "port": self.metrics_port,
        }

    doc: dict[str, Any] = {
        "logLevel": self.log_level,
        "server": server_dict,
        "metrics": metrics_dict,
        "projects": [self._build_project()],
    }
    return yaml.dump(doc, default_flow_style=False, sort_keys=False)

write 

write(path: Path | None = None) -> Path

Write config to a YAML file.

Parameters:

Name Type Description Default
path Path | None

Destination file path. If None, writes to a temporary file.

 None

Returns:

Type Description
Path

Path to the written configuration file.
Source code in erpc/config.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def write(self, path: Path | None = None) -> Path:
    """Write config to a YAML file.

    Args:
        path: Destination file path. If ``None``, writes to a temporary file.

    Returns:
        Path to the written configuration file.

    """
    if path is None:
        fd = tempfile.NamedTemporaryFile(
            mode="w", suffix=".yaml", prefix="erpc-", delete=False
        )
        fd.write(self.to_yaml())
        fd.close()
        return Path(fd.name)
    else:
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.to_yaml())
        return path