RPCProxy

Documentation for eth_defi.provider.rpc_proxy.RPCProxy Python class.

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

  1. start_rpc_proxy() allocates a free localhost port and starts a HTTPServer (with ThreadingMixIn) on a daemon thread.

  2. 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.

  3. Connection-level failures (timeouts, refused connections) are always retried without consulting the failure handler.

  4. 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.

  5. 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.

See also RPCProxyConfig, start_rpc_proxy(), UpstreamRPCProviderStatistics.

Attributes summary

name

port

url

provider_stats

Methods summary

__init__(name, port, url, provider_stats, ...)

close()

Shut down the proxy server and log final statistics.

get_stats()

Return per-provider statistics, keyed by display URL.
__init__(name, port, url, provider_stats, _server_thread, _http_server)
Parameters
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]