provider.rpc_proxy

Documentation for eth_defi.provider.rpc_proxy Python module.

JSON-RPC failover proxy for multiple upstream RPC providers.

A lightweight threaded HTTP proxy that presents a single JSON-RPC endpoint while internally routing requests across multiple upstream RPC providers with automatic failover, retry, and per-provider statistics.

The primary use case is Anvil mainnet forks: Anvil accepts only a single --fork-url and has no internal retry or failover logic. When the upstream RPC is slow or rate-limited, Anvil hangs indefinitely. This proxy sits between Anvil and the upstream RPCs, transparently handling failures.

However the proxy is general-purpose and can be used with any software that needs a single RPC URL backed by multiple upstreams.

Example usage:

from eth_defi.provider.rpc_proxy import start_rpc_proxy

proxy = start_rpc_proxy(
    [
        "https://rpc-provider-a.example.com",
        "https://rpc-provider-b.example.com",
    ]
)
print(f"Proxy listening at {proxy.url}")

# Pass proxy.url to Anvil, or any other JSON-RPC client
# ...

# When done, shut down and see statistics
proxy.close()

See eth_defi.provider.anvil.launch_anvil() for automatic integration.

Functions

`default_failure_handler`(http_status, json_body)	Default failure detection replicating `eth_defi.middleware` logic.
`start_rpc_proxy`(rpc_urls[, port, config])	Start a JSON-RPC failover proxy on a background thread.

Classes

`RPCProxy`	A running JSON-RPC failover proxy instance.
`RPCProxyConfig`	Configuration for the JSON-RPC failover proxy.
`UpstreamRPCProviderStatistics`	Per-provider statistics collected during proxy operation.

class RPCProxy

Bases: object

A running JSON-RPC failover proxy instance.

Manages the lifecycle of a background threaded HTTP server that presents a single http://127.0.0.1:{port} endpoint while internally routing JSON-RPC requests across multiple upstream RPC providers with automatic failover, retry, and per-provider statistics collection.

Why this exists

Anvil (and other tools) accept only a single --fork-url and have no internal retry or failover logic. When the upstream RPC is slow, rate-limited, or temporarily unreachable, Anvil hangs indefinitely — causing downstream callers to timeout (e.g. eth_getTransactionCount timing out after 90 s).

This proxy sits between the consumer and multiple upstream RPCs. If one upstream fails, the proxy transparently switches to the next one and retries, all within a configurable timeout budget. On shutdown it logs per-provider statistics so you can identify flaky or slow providers.

How it works

start_rpc_proxy() allocates a free localhost port and starts a HTTPServer (with ThreadingMixIn) on a daemon thread.
Every incoming POST is forwarded to the currently-active upstream. If the upstream returns a retryable error (as determined by the FailureHandler), the proxy switches to the next upstream and retries — up to DEFAULT_RETRIES times.
Connection-level failures (timeouts, refused connections) are always retried without consulting the failure handler.
After all retries are exhausted the proxy returns an HTTP 502 with a JSON-RPC error body so the caller can distinguish proxy-level failures from upstream errors.
Calling close() stops the server and logs a per-provider summary to logger.info().

Lifecycle

Created by start_rpc_proxy(). The proxy runs until close() is called. When used with launch_anvil(), the lifecycle is automatic: the proxy starts before Anvil and shuts down when close() is called.

Standalone usage

from eth_defi.provider.rpc_proxy import start_rpc_proxy

# Start a proxy backed by two upstream RPCs
proxy = start_rpc_proxy(
    [
        "https://eth-mainnet.alchemyapi.io/v2/YOUR_KEY",
        "https://rpc.ankr.com/eth",
    ]
)

# proxy.url is e.g. "http://127.0.0.1:23456"
# Pass it to any tool that accepts a single JSON-RPC URL:
#   anvil --fork-url {proxy.url}
#   curl -X POST {proxy.url} -d '{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":1}'

# Inspect statistics at any time
for name, stats in proxy.get_stats().items():
    print(f"{name}: {stats.request_count} requests, {stats.failure_count} failures")

# Shut down and see final statistics in the log
proxy.close()

With custom parameters

import logging
from eth_defi.provider.rpc_proxy import start_rpc_proxy

proxy = start_rpc_proxy(
    rpc_urls=[
        "https://rpc-provider-a.example.com",
        "https://rpc-provider-b.example.com",
    ],
    timeout=15.0,  # 15 s per upstream attempt
    retries=5,  # try up to 5 times
    auto_switch_request_count=100,  # rotate providers every 100 requests
    switchover_log_level=logging.WARNING,
    request_log_level=logging.DEBUG,  # dump payloads at DEBUG level
    log_max_size=4096,  # truncate large payloads at 4 KiB
)

Automatic integration with Anvil

When launch_anvil() receives a space-separated fork_url containing multiple RPC endpoints, it automatically starts an RPCProxy and passes proxy.url as Anvil’s --fork-url. The proxy’s lifecycle is tied to close():

from eth_defi.provider.anvil import launch_anvil

# Space-separated URLs trigger the proxy automatically
launch = launch_anvil(
    fork_url="https://rpc-a.example.com https://rpc-b.example.com",
)
# launch.proxy is the RPCProxy instance
# ...run your test...
launch.close()  # stops Anvil, then stops the proxy and logs stats

You can also pass an RPCProxy you created yourself, or an RPCProxyConfig to fine-tune settings, via the proxy_multiple_upstream parameter — see launch_anvil() for details.

__init__(name, port, url, provider_stats, _server_thread, _http_server)

Parameters

name (str) –
port (int) –
url (str) –
provider_stats (dict[str, eth_defi.provider.rpc_proxy.UpstreamRPCProviderStatistics]) –
_server_thread (threading.Thread) –
_http_server (eth_defi.provider.rpc_proxy._ThreadingHTTPServer) –

Return type

None

close()

Shut down the proxy server and log final statistics.

Stops accepting new requests, waits for the server thread to finish, then logs a summary of per-provider statistics at INFO level.

Return type: None

get_stats()

Return per-provider statistics, keyed by display URL.

Returns: Dictionary mapping provider display name to its statistics.
Return type: dict[str, eth_defi.provider.rpc_proxy.UpstreamRPCProviderStatistics]

class RPCProxyConfig

Bases: object

Configuration for the JSON-RPC failover proxy.

Collects all tuneable parameters of start_rpc_proxy() into a single object with sensible defaults. Every field has a docstring explaining its purpose, default value, and interaction with other fields.

You can construct this directly and pass it to start_rpc_proxy(), or pass it as the proxy_multiple_upstream argument to launch_anvil().

Example — standalone proxy with custom configuration:

from eth_defi.provider.rpc_proxy import RPCProxyConfig, start_rpc_proxy

config = RPCProxyConfig(
    timeout=15.0,
    retries=5,
    auto_switch_request_count=50,
)
proxy = start_rpc_proxy(
    ["https://rpc-a.example.com", "https://rpc-b.example.com"],
    config=config,
)

Example — passing to launch_anvil:

from eth_defi.provider.anvil import launch_anvil
from eth_defi.provider.rpc_proxy import RPCProxyConfig

config = RPCProxyConfig(timeout=10.0, retries=4)
launch = launch_anvil(
    fork_url="https://rpc-a.example.com https://rpc-b.example.com",
    proxy_multiple_upstream=config,
)

__init__(name=None, timeout=30.0, retries=3, backoff=0.5, auto_switch_request_count=0, switchover_log_level=20, request_log_level=10, log_max_size=2048, pool_maxsize=50, max_error_replies=100, failure_handler=None)

Parameters

name (Optional[str]) –
timeout (float) –
retries (int) –
backoff (float) –
auto_switch_request_count (int) –
switchover_log_level (int) –
request_log_level (int) –
log_max_size (int) –
pool_maxsize (int) –
max_error_replies (int) –
failure_handler (Callable[[int, Optional[dict]], bool]) –

Return type

None

describe()

Return a human-readable summary of the configuration for logging.

Includes all tuneable numeric fields for full visibility.

Example output:

timeout=30.0, retries=3, backoff=0.5, auto_switch_request_count=0, pool_maxsize=50, max_error_replies=100, log_max_size=2048

Return type: str

class UpstreamRPCProviderStatistics

Bases: object

Per-provider statistics collected during proxy operation.

Tracks request counts, failure counts, and method-level breakdowns for each upstream RPC provider. Instances are keyed by provider URL in RPCProxy.provider_stats.

__init__(url, request_count=0, failure_count=0, last_failure=None, method_counts=<factory>, method_failure_counts=<factory>, error_replies=<factory>, _lock=<factory>)

Parameters

url (str) –
request_count (int) –
failure_count (int) –
last_failure (Optional[datetime.datetime]) –
method_counts (dict[str, int]) –
method_failure_counts (dict[str, int]) –
error_replies (list[dict]) –
_lock (_thread.lock) –

Return type

None

record_failure(method, error_summary, http_status=None, max_error_replies=100)

Record a failed request to this provider.

Parameters

method (str) –
error_summary (str) –
http_status (Optional[int]) –
max_error_replies (int) –

Return type

None

record_request(method)

Record a request being sent to this provider.

Parameters: method (str) –
Return type: None

default_failure_handler(http_status, json_body)

Default failure detection replicating eth_defi.middleware logic.

Adapted from eth_defi.middleware.is_retryable_http_exception() and eth_defi.provider.fallback.FallbackProvider to work at the HTTP proxy level with raw status codes and JSON bodies rather than Python exceptions.

The checks are, in order:

HTTP status code against eth_defi.middleware.DEFAULT_RETRYABLE_HTTP_STATUS_CODES
JSON-RPC error.code against eth_defi.middleware.DEFAULT_RETRYABLE_RPC_ERROR_CODES
JSON-RPC error.message substring match against eth_defi.middleware.DEFAULT_RETRYABLE_RPC_ERROR_MESSAGES

Connection-level failures (timeouts, refused connections) are always retried and never reach this handler — they are caught earlier in _ProxyRequestHandler._try_upstream().

Parameters

http_status (int) – HTTP response status code from the upstream provider.
json_body (Optional[dict]) – Parsed JSON-RPC response body, or None if unparseable.

Returns

True if the response should be treated as a retryable failure.

Return type

bool

start_rpc_proxy(rpc_urls, port=None, config=None, **kwargs)

Start a JSON-RPC failover proxy on a background thread.

The proxy listens on localhost and forwards incoming JSON-RPC POST requests to the given upstream RPC URLs with automatic failover, retry, and statistics collection.

Parameters

rpc_urls (list[str]) – Upstream RPC endpoint URLs to cycle through. At least one URL is required.
port (Optional[int]) –
Local port to bind on 127.0.0.1.

If None (the default), the server binds to port 0 which tells the operating system to assign a free ephemeral port atomically. The actual port is read back from the socket after binding and stored in RPCProxy.port. This avoids the TOCTOU race condition that occurs with find_free_port(): that function checks availability via connect(), but under heavy parallel test execution (pytest -n auto) another process can grab the same port between the check and the bind() call, resulting in OSError: [Errno 98] Address already in use.

Pass an explicit port number only when you need a deterministic address (e.g. for debugging or firewall rules).
config (Optional[eth_defi.provider.rpc_proxy.RPCProxyConfig]) – Proxy configuration. If None, a default RPCProxyConfig is used. Individual fields can be overridden via **kwargs.
kwargs – Override individual RPCProxyConfig fields. For example start_rpc_proxy(urls, timeout=10.0) is equivalent to start_rpc_proxy(urls, config=RPCProxyConfig(timeout=10.0)). When both config and kwargs are provided, kwargs win.

Returns

A running RPCProxy instance. Call RPCProxy.close() when done.

Return type

eth_defi.provider.rpc_proxy.RPCProxy