tessl/pypi-web3

A Python library for interacting with Ethereum blockchain

Overall
score

88%

Evaluation — 88%

↑ 1.01x

Agent success when using this tile

Overview

Eval results

Files

Providers

Name: tessl/pypi-web3
Author: tessl

Connection providers for different Ethereum client interfaces supporting HTTP, WebSocket, IPC, and testing environments with automatic detection and persistent connection management.

Capabilities

HTTP Providers

JSON-RPC over HTTP connection providers for standard Ethereum node communication.

class HTTPProvider(JSONBaseProvider):
    def __init__(
        self,
        endpoint_uri: Optional[Union[URI, str]] = None,
        request_kwargs: Optional[Dict[str, Any]] = None
    ):
        """
        HTTP JSON-RPC provider.
        
        Parameters:
        - endpoint_uri: HTTP endpoint URL
        - request_kwargs: Additional kwargs for requests
        """

class AsyncHTTPProvider(AsyncJSONBaseProvider):
    def __init__(
        self,
        endpoint_uri: Optional[Union[URI, str]] = None,
        request_kwargs: Optional[Dict[str, Any]] = None
    ):
        """
        Async HTTP JSON-RPC provider.
        
        Parameters:
        - endpoint_uri: HTTP endpoint URL  
        - request_kwargs: Additional kwargs for aiohttp requests
        """

WebSocket Providers

WebSocket connection providers supporting both legacy and persistent connection patterns.

class WebSocketProvider(PersistentConnectionProvider):
    def __init__(
        self,
        endpoint_uri: Optional[Union[URI, str]] = None,
        websockets_kwargs: Optional[Dict[str, Any]] = None
    ):
        """
        WebSocket provider with persistent connection.
        
        Parameters:
        - endpoint_uri: WebSocket endpoint URL
        - websockets_kwargs: Additional kwargs for websockets connection
        """

class LegacyWebSocketProvider(JSONBaseProvider):
    def __init__(
        self,
        endpoint_uri: Optional[Union[URI, str]] = None,
        websockets_kwargs: Optional[Dict[str, Any]] = None
    ):
        """
        Legacy WebSocket provider.
        
        Parameters:
        - endpoint_uri: WebSocket endpoint URL
        - websockets_kwargs: Additional kwargs for websockets connection
        """

IPC Providers

Inter-Process Communication providers for local Ethereum node connections.

class IPCProvider(JSONBaseProvider):
    def __init__(self, ipc_path: Optional[str] = None):
        """
        IPC provider for Unix socket connections.
        
        Parameters:
        - ipc_path: Path to IPC socket file
        """

class AsyncIPCProvider(PersistentConnectionProvider):
    def __init__(self, ipc_path: Optional[str] = None):
        """
        Async IPC provider with persistent connection.
        
        Parameters:
        - ipc_path: Path to IPC socket file
        """

Testing Providers

Providers for testing environments with in-memory blockchain simulation.

class EthereumTesterProvider(BaseProvider):
    def __init__(self, eth_tester: Optional[EthereumTester] = None):
        """
        Provider for eth-tester backend.
        
        Parameters:
        - eth_tester: EthereumTester instance
        """

class AsyncEthereumTesterProvider(AsyncBaseProvider):
    def __init__(self, eth_tester: Optional[EthereumTester] = None):
        """
        Async provider for eth-tester backend.
        
        Parameters:
        - eth_tester: EthereumTester instance
        """

Auto Provider

Automatic provider detection and configuration.

class AutoProvider:
    @staticmethod
    def from_auto() -> BaseProvider:
        """
        Automatically detect and configure provider.
        
        Returns:
        Configured provider based on environment
        """

Base Provider Classes

Abstract base classes defining provider interfaces.

class BaseProvider:
    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
        """Make JSON-RPC request."""

    def is_connected(self) -> bool:
        """Check if provider is connected."""

class AsyncBaseProvider:
    async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
        """Make async JSON-RPC request."""

    async def is_connected(self) -> bool:
        """Check if provider is connected."""

class JSONBaseProvider(BaseProvider):
    def decode_rpc_response(self, response: bytes) -> RPCResponse:
        """Decode JSON-RPC response."""

    def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
        """Encode JSON-RPC request."""

class PersistentConnectionProvider(AsyncBaseProvider):
    def __init__(self, endpoint_uri: Optional[str] = None):
        """Base class for persistent connection providers."""

    async def connect(self) -> None:
        """Establish persistent connection."""

    async def disconnect(self) -> None:
        """Close persistent connection."""

Connection Management

class PersistentConnection:
    def __init__(self, provider: PersistentConnectionProvider):
        """Manage persistent connection lifecycle."""

    async def __aenter__(self) -> PersistentConnection:
        """Async context manager entry."""

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:  
        """Async context manager exit."""

Usage Examples

HTTP Provider

from web3 import Web3

# Mainnet via Infura
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR-PROJECT-ID'))

# Local node
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Custom request configuration
provider = Web3.HTTPProvider(
    'https://mainnet.infura.io/v3/YOUR-PROJECT-ID',
    request_kwargs={'timeout': 60}
)
w3 = Web3(provider)

WebSocket Provider

from web3 import Web3

# WebSocket connection with subscription support
w3 = Web3(Web3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID'))

# Local WebSocket
w3 = Web3(Web3.WebSocketProvider('ws://localhost:8546'))

IPC Provider

from web3 import Web3

# Default IPC path (varies by OS)
w3 = Web3(Web3.IPCProvider())

# Custom IPC path
w3 = Web3(Web3.IPCProvider('/path/to/geth.ipc'))

Testing Provider

from web3 import Web3
from eth_tester import EthereumTester

# Use eth-tester for testing
w3 = Web3(Web3.EthereumTesterProvider())

# Custom eth-tester configuration
eth_tester = EthereumTester()
w3 = Web3(Web3.EthereumTesterProvider(eth_tester))

Auto Provider Detection

from web3 import Web3

# Automatically detect provider
w3 = Web3(Web3.AutoProvider.from_auto())

Async Providers

import asyncio
from web3 import AsyncWeb3

async def main():
    # Async HTTP
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider('https://mainnet.infura.io/v3/YOUR-PROJECT-ID'))
    
    # Async WebSocket with persistent connection
    w3 = AsyncWeb3(AsyncWeb3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID'))
    
    # Check connection
    connected = await w3.is_connected()
    print(f"Connected: {connected}")

asyncio.run(main())

Persistent Connection Management

import asyncio
from web3 import AsyncWeb3

async def main():
    provider = AsyncWeb3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID')
    w3 = AsyncWeb3(provider)
    
    async with provider:
        # Connection automatically managed
        block = await w3.eth.get_block('latest')
        print(f"Block number: {block.number}")

asyncio.run(main())

Install with Tessl CLI

npx tessl i tessl/pypi-web3