How to Connect to Ethereum with web3.py: RPC Configuration Guide

web3.py is the standard Python library for Ethereum. If you are building data pipelines, analytics scripts, trading bots, or backend services in Python, web3.py is how you connect to the blockchain.

By default, tutorials point you at public endpoints that get rate-limited immediately under any real workload. This guide shows how to configure web3.py properly with a dedicated RPC endpoint.

---

Install web3.py

pip install web3

Requires Python 3.7+. For async support, install the extras:

pip install "web3[asyncio]"

---

HTTP Provider

The HTTP provider handles all standard JSON-RPC calls: reading blocks, balances, contract state.

from web3 import Web3

w3 = Web3(Web3.HTTPProvider(
    'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY'
))

# Verify connection
print(w3.is_connected())  # True

# Latest block
block = w3.eth.get_block('latest')
print(f"Latest block: {block.number}")

# ETH balance
balance = w3.eth.get_balance('0xYourAddress')
print(f"Balance: {w3.from_wei(balance, 'ether')} ETH")

---

WebSocket Provider

For event subscriptions and real-time data, use the WebSocket provider:

from web3 import Web3

w3 = Web3(Web3.WebsocketProvider(
    'wss://eu.endpoints.matrixed.link/ws/ethereum?auth=YOUR_KEY'
))

print(w3.is_connected())  # True

WebSocket is required for eth_subscribe calls (new blocks, pending transactions, contract logs).

---

Async HTTP Provider

For production scripts that need concurrent requests:

import asyncio
from web3 import AsyncWeb3

async def main():
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(
        'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY'
    ))

    block = await w3.eth.get_block('latest')
    print(f"Latest block: {block.number}")

    # Concurrent requests
    block_num, gas_price, chain_id = await asyncio.gather(
        w3.eth.block_number,
        w3.eth.gas_price,
        w3.eth.chain_id,
    )
    print(f"Block: {block_num}, Gas: {gas_price}, Chain: {chain_id}")

asyncio.run(main())

---

Request Session Configuration

For production use, configure connection pooling and timeouts:

from web3 import Web3
from web3.middleware import ExtraDataToPOAMiddleware
from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure retry + connection pool
session = Session()
adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=[429, 500, 502, 503, 504],
    )
)
session.mount('https://', adapter)

w3 = Web3(Web3.HTTPProvider(
    'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY',
    request_kwargs={
        'timeout': 30,
        'session': session,
    }
))

For POA chains (Polygon, BNB Chain), inject the POA middleware:

# Required for Polygon, BNB Chain, and other POA networks
w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)

---

Read Contract Data

from web3 import Web3

w3 = Web3(Web3.HTTPProvider(
    'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY'
))

ERC20_ABI = [
    {
        "inputs": [{"name": "account", "type": "address"}],
        "name": "balanceOf",
        "outputs": [{"name": "", "type": "uint256"}],
        "stateMutability": "view",
        "type": "function",
    },
    {
        "inputs": [],
        "name": "decimals",
        "outputs": [{"name": "", "type": "uint8"}],
        "stateMutability": "view",
        "type": "function",
    },
    {
        "inputs": [],
        "name": "symbol",
        "outputs": [{"name": "", "type": "string"}],
        "stateMutability": "view",
        "type": "function",
    },
]

USDC_ADDRESS = Web3.to_checksum_address('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48')
contract = w3.eth.contract(address=USDC_ADDRESS, abi=ERC20_ABI)

# Read contract state
symbol = contract.functions.symbol().call()
decimals = contract.functions.decimals().call()
balance = contract.functions.balanceOf('0xYourAddress').call()

print(f"{symbol}: {balance / 10**decimals}")

---

Send a Transaction

from web3 import Web3

w3 = Web3(Web3.HTTPProvider(
    'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY'
))

private_key = 'YOUR_PRIVATE_KEY'
account = w3.eth.account.from_key(private_key)

# Build transaction
nonce = w3.eth.get_transaction_count(account.address)
fee_data = w3.eth.fee_history(1, 'latest', [50])
base_fee = fee_data['baseFeePerGas'][-1]
priority_fee = w3.to_wei(1, 'gwei')

tx = {
    'chainId': 1,
    'from': account.address,
    'to': '0xRecipientAddress',
    'value': w3.to_wei(0.001, 'ether'),
    'gas': 21000,
    'maxFeePerGas': base_fee * 2 + priority_fee,
    'maxPriorityFeePerGas': priority_fee,
    'nonce': nonce,
    'type': 2,  # EIP-1559
}

# Sign and send
signed = account.sign_transaction(tx)
tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"Confirmed in block {receipt.blockNumber}")

---

Filter Events with eth_getLogs

from web3 import Web3

w3 = Web3(Web3.HTTPProvider(
    'https://eu.endpoints.matrixed.link/rpc/ethereum?auth=YOUR_KEY'
))

TRANSFER_TOPIC = Web3.keccak(text='Transfer(address,address,uint256)').hex()
USDC_ADDRESS = '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'

current_block = w3.eth.block_number

# Get USDC transfers in the last 1,000 blocks
logs = w3.eth.get_logs({
    'address': Web3.to_checksum_address(USDC_ADDRESS),
    'topics': [TRANSFER_TOPIC],
    'fromBlock': current_block - 1000,
    'toBlock': 'latest',
})

print(f"Found {len(logs)} USDC Transfer events")
for log in logs[:5]:
    print(f"  TX: {log.transactionHash.hex()}, Block: {log.blockNumber}")

For large block ranges, chunk requests to avoid provider limits:

def get_logs_chunked(w3, filter_params, chunk_size=2000):
    all_logs = []
    from_block = filter_params['fromBlock']
    to_block = filter_params['toBlock']

    current = from_block
    while current <= to_block:
        end = min(current + chunk_size - 1, to_block)
        try:
            logs = w3.eth.get_logs({**filter_params, 'fromBlock': current, 'toBlock': end})
            all_logs.extend(logs)
            current = end + 1
        except Exception as e:
            if 'block range' in str(e).lower() or 'too many' in str(e).lower():
                chunk_size = chunk_size // 2
                if chunk_size < 10:
                    raise
                continue
            raise
    return all_logs

---

WebSocket Subscriptions

import asyncio
from web3 import AsyncWeb3

async def watch_new_blocks():
    w3 = AsyncWeb3(AsyncWeb3.AsyncWebsocketProvider(
        'wss://eu.endpoints.matrixed.link/ws/ethereum?auth=YOUR_KEY'
    ))

    subscription_id = await w3.eth.subscribe('newHeads')
    print(f"Subscribed to new blocks: {subscription_id}")

    async for response in w3.socket.process_subscriptions():
        block = response['result']
        print(f"New block: {block['number']} | Hash: {block['hash'].hex()}")

asyncio.run(watch_new_blocks())

Subscribe to contract logs in real time:

async def watch_usdc_transfers():
    w3 = AsyncWeb3(AsyncWeb3.AsyncWebsocketProvider(
        'wss://eu.endpoints.matrixed.link/ws/ethereum?auth=YOUR_KEY'
    ))

    TRANSFER_TOPIC = '0x' + Web3.keccak(text='Transfer(address,address,uint256)').hex()

    subscription_id = await w3.eth.subscribe('logs', {
        'address': USDC_ADDRESS,
        'topics': [TRANSFER_TOPIC],
    })

    async for response in w3.socket.process_subscriptions():
        log = response['result']
        print(f"USDC Transfer — TX: {log['transactionHash'].hex()}")

---

Multi-Chain Configuration

from web3 import Web3

RPC_BASE = 'https://eu.endpoints.matrixed.link/rpc'
AUTH = 'YOUR_KEY'

chains = {
    'ethereum': Web3(Web3.HTTPProvider(f'{RPC_BASE}/ethereum?auth={AUTH}')),
    'polygon':  Web3(Web3.HTTPProvider(f'{RPC_BASE}/polygon?auth={AUTH}')),
    'arbitrum': Web3(Web3.HTTPProvider(f'{RPC_BASE}/arbitrum?auth={AUTH}')),
    'base':     Web3(Web3.HTTPProvider(f'{RPC_BASE}/base?auth={AUTH}')),
}

# Apply POA middleware to POA chains
from web3.middleware import ExtraDataToPOAMiddleware
chains['polygon'].middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)

# Query any chain
for name, w3 in chains.items():
    block = w3.eth.block_number
    print(f"{name}: block {block}")

---

Environment Variables

import os
from web3 import Web3

# Load from environment, never hardcode
BOLTRPC_KEY = os.environ['BOLTRPC_KEY']

w3 = Web3(Web3.HTTPProvider(
    f'https://eu.endpoints.matrixed.link/rpc/ethereum?auth={BOLTRPC_KEY}'
))

With python-dotenv:

pip install python-dotenv

from dotenv import load_dotenv
import os
from web3 import Web3

load_dotenv()

w3 = Web3(Web3.HTTPProvider(
    f'https://eu.endpoints.matrixed.link/rpc/ethereum?auth={os.environ["BOLTRPC_KEY"]}'
))

---

FAQ

Does web3.py work with all EVM chains?

Yes. web3.py connects to any EVM-compatible JSON-RPC endpoint: Ethereum, Polygon, Arbitrum, Base, Optimism, BNB Chain, Avalanche, and others. For POA chains (Polygon, BNB Chain), inject ExtraDataToPOAMiddleware as shown above.

What is the difference between Web3.HTTPProvider and Web3.WebsocketProvider?

HTTP is request-response: you send a query and get a response. WebSocket is persistent and bidirectional: you can subscribe to events and receive data as it arrives. Use HTTP for reading data; use WebSocket when you need real-time subscriptions to new blocks or contract events.

How do I handle rate limit errors in web3.py?

Configure HTTPAdapter with Retry as shown above. Set status_forcelist=[429, 500, 502, 503, 504] and backoff_factor=0.5 for exponential backoff. On a dedicated RPC plan with flat-rate pricing, rate limits are far less likely than on public endpoints.

Is web3.py thread-safe?

The HTTPProvider can be used across threads if you configure connection pooling. The WebsocketProvider is not thread-safe. Use asyncio instead. For multi-threaded scripts, create one Web3 instance per thread or use a connection pool.

Can I use web3.py with Django or FastAPI?

Yes. For Django (synchronous), use Web3(Web3.HTTPProvider(...)) directly. For FastAPI (async), use AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(...)) and await calls from async route handlers. Avoid blocking web3.py calls inside async handlers. They will block the event loop.

---

BoltRPC provides HTTP and WebSocket endpoints for Ethereum, Polygon, Arbitrum, Base and 17 other networks, all accessible with the same web3.py configuration pattern.

Start your free 2-week trial: trial.boltrpc.io

Related: How to Connect Ethereum with ethers.js | Optimizing RPC Calls for DeFi