AsyncShieldedContract

Async contract wrapper with shielded and transparent namespaces

Asynchronous contract wrapper providing encrypted and transparent interaction with Seismic contracts.

Overview

AsyncShieldedContract is the async version of ShieldedContract, providing the same five namespaces (.write, .read, .twrite, .tread, .dwrite) but with coroutine-based methods. All namespace methods return coroutines that must be awaited. Use this class in async/await applications for non-blocking contract interactions.

Definition

class AsyncShieldedContract:
    def __init__(
        self,
        w3: AsyncWeb3,
        encryption: EncryptionState,
        private_key: PrivateKey,
        address: ChecksumAddress,
        abi: list[dict[str, Any]],
        eip712: bool = False,
    ) -> None:
        ...

Constructor Parameters

Parameter

Type

Required

Description

w3

AsyncWeb3

Yes

Asynchronous AsyncWeb3 instance connected to Seismic RPC

encryption

EncryptionState

Yes

Encryption state for shielded operations

private_key

PrivateKey

Yes

32-byte secp256k1 private key for signing transactions

address

ChecksumAddress

Yes

Contract address (checksummed Ethereum address)

abi

list[dict[str, Any]]

Yes

Contract ABI (list of function entries)

eip712

bool

Use EIP-712 typed data signing (default: False)

Namespaces

`.write` - Encrypted Write

Sends encrypted transactions using TxSeismic (type 0x4a). Calldata is encrypted before broadcast.

Returns: Coroutine[HexBytes] (transaction hash)

Optional Parameters:

value: int - Wei to send (default: 0)
gas: int | None - Gas limit (default: estimated)
gas_price: int | None - Gas price in wei (default: network suggested)
security: SeismicSecurityParams | None - Security parameters for expiry

`.read` - Encrypted Read

Executes encrypted signed eth_call with encrypted calldata. Result is decrypted by the SDK.

Returns: Coroutine[HexBytes] (decrypted result bytes)

Optional Parameters:

value: int - Wei for call context (default: 0)
gas: int - Gas limit (default: 30_000_000)
security: SeismicSecurityParams | None - Security parameters for expiry

`.twrite` - Transparent Write

Sends standard async eth_sendTransaction with unencrypted calldata.

Returns: Coroutine[HexBytes] (transaction hash)

Optional Parameters:

value: int - Wei to send (default: 0)
**tx_params: Any - Additional transaction parameters (gas, gasPrice, etc.)

`.tread` - Transparent Read

Executes standard async eth_call with unencrypted calldata.

Returns: Coroutine[HexBytes] (raw result bytes)

Optional Parameters: None (pass positional arguments only)

`.dwrite` - Debug Write

Like .write but returns debug information including plaintext and encrypted views. Transaction is actually broadcast.

Returns: Coroutine[DebugWriteResult] (DebugWriteResult)

Optional Parameters: Same as .write

Examples

Basic Encrypted Write

import asyncio
from seismic_web3 import create_async_wallet_client, AsyncShieldedContract, SEISMIC_TESTNET

async def main():
    w3 = await create_async_wallet_client(
        rpc_url="https://gcp-1.seismictest.net/rpc",
        chain=SEISMIC_TESTNET,
        account=private_key,
    )

    contract = AsyncShieldedContract(
        w3=w3.eth,
        encryption=w3.seismic.encryption,
        private_key=private_key,
        address="0x1234567890123456789012345678901234567890",
        abi=CONTRACT_ABI,
    )

    # Encrypted write - must await
    tx_hash = await contract.write.setNumber(42)
    print(f"Transaction: {tx_hash.to_0x_hex()}")

    # Wait for confirmation
    receipt = await w3.eth.wait_for_transaction_receipt(tx_hash)
    print(f"Status: {receipt['status']}")

asyncio.run(main())

Encrypted Read

async def read_example(contract: AsyncShieldedContract):
    # Encrypted read - must await
    result = await contract.read.getNumber()

    if result:
        print(f"Raw result: {result.to_0x_hex()}")

Transparent Operations

async def transparent_example(contract: AsyncShieldedContract, w3: AsyncWeb3):
    # Transparent write - calldata visible on-chain
    tx_hash = await contract.twrite.setPublicData("hello", value=10**18)
    receipt = await w3.eth.wait_for_transaction_receipt(tx_hash)
    print(f"Status: {receipt['status']}")

    # Transparent read - standard eth_call
    result = await contract.tread.getPublicData()
    print(f"Result: {result.to_0x_hex()}")

Debug Write

async def debug_example(contract: AsyncShieldedContract, w3: AsyncWeb3):
    # Debug write - returns plaintext and encrypted views
    debug_result = await contract.dwrite.transfer("0xRecipient...", 1000)

    print(f"Transaction hash: {debug_result.tx_hash.to_0x_hex()}")
    print(f"Plaintext data: {debug_result.plaintext_tx.data.to_0x_hex()}")
    print(f"Encrypted data: {debug_result.shielded_tx.data.to_0x_hex()}")

    # Transaction is actually broadcast
    receipt = await w3.eth.wait_for_transaction_receipt(debug_result.tx_hash)
    print(f"Confirmed in block: {receipt['blockNumber']}")

Concurrent Operations

async def concurrent_example(contract: AsyncShieldedContract):
    # Execute multiple reads concurrently
    results = await asyncio.gather(
        contract.tread.getBalance("0xAddress1..."),
        contract.tread.getBalance("0xAddress2..."),
        contract.tread.getBalance("0xAddress3..."),
    )

    for i, result in enumerate(results):
        print(f"Balance {i}: {result.to_0x_hex()}")

With Transaction Parameters

async def advanced_write(contract: AsyncShieldedContract):
    # Custom gas and value
    tx_hash = await contract.write.deposit(
        amount=1000,
        value=10**18,  # 1 ETH
        gas=200_000,
        gas_price=20 * 10**9,  # 20 gwei
    )

    # With security parameters
    from seismic_web3.transaction_types import SeismicSecurityParams

    security = SeismicSecurityParams(expires_in_blocks=100)
    tx_hash = await contract.write.sensitiveOperation(
        data="secret",
        security=security,
    )

Batch Processing

async def batch_writes(contract: AsyncShieldedContract, recipients: list[str]):
    # Send multiple transactions concurrently
    tx_hashes = await asyncio.gather(
        *[contract.write.transfer(recipient, 100) for recipient in recipients]
    )

    print(f"Sent {len(tx_hashes)} transactions")

    # Wait for all confirmations
    receipts = await asyncio.gather(
        *[w3.eth.wait_for_transaction_receipt(tx_hash) for tx_hash in tx_hashes]
    )

    successful = sum(1 for r in receipts if r['status'] == 1)
    print(f"{successful}/{len(receipts)} successful")

Using EIP-712 Signing

async def eip712_example():
    w3 = await create_async_wallet_client(...)

    # Enable EIP-712 for typed data signing
    contract = AsyncShieldedContract(
        w3=w3.eth,
        encryption=w3.seismic.encryption,
        private_key=private_key,
        address=contract_address,
        abi=CONTRACT_ABI,
        eip712=True,  # Use EIP-712 instead of raw signing
    )

    tx_hash = await contract.write.performAction(123)

Instantiation via Async Client

async def client_pattern():
    # Most common pattern - let the client create the contract
    w3 = await create_async_wallet_client(
        rpc_url="https://gcp-1.seismictest.net/rpc",
        chain=SEISMIC_TESTNET,
        account=private_key,
    )

    # Client's contract() method creates AsyncShieldedContract
    contract = w3.seismic.contract(address=contract_address, abi=CONTRACT_ABI)

    # Now use any namespace (must await)
    tx_hash = await contract.write.myFunction(arg1, arg2)

Error Handling

async def error_handling(contract: AsyncShieldedContract):
    try:
        tx_hash = await contract.write.riskyOperation(123)
        receipt = await w3.eth.wait_for_transaction_receipt(tx_hash)

        if receipt['status'] != 1:
            print("Transaction failed on-chain")
    except ValueError as e:
        print(f"RPC error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

Context Manager Pattern

async def context_pattern():
    async with create_async_wallet_client(...) as w3:
        contract = AsyncShieldedContract(...)

        tx_hash = await contract.write.doSomething()
        receipt = await w3.eth.wait_for_transaction_receipt(tx_hash)
    # Connection automatically closed

Notes

All methods return coroutines: Must use await with every namespace call
Concurrent operations: Use asyncio.gather() for parallel reads/writes
Dynamic method access: Contract methods resolved via __getattr__ at runtime
ABI remapping: Shielded types automatically remapped (same as sync version)
Connection pooling: AsyncWeb3 can reuse connections for better performance
Gas estimation: Happens asynchronously if not provided
EIP-712 vs raw: Same signing options as sync version
Error handling: Use try/except around await calls for RPC errors

hashtagOverview

hashtagDefinition

hashtagConstructor Parameters

hashtagNamespaces

hashtag.write - Encrypted Write

hashtag.read - Encrypted Read

hashtag.twrite - Transparent Write

hashtag.tread - Transparent Read

hashtag.dwrite - Debug Write

hashtagExamples

hashtagBasic Encrypted Write

hashtagEncrypted Read

hashtagTransparent Operations

hashtagDebug Write

hashtagConcurrent Operations

hashtagWith Transaction Parameters

hashtagBatch Processing

hashtagUsing EIP-712 Signing

hashtagInstantiation via Async Client

hashtagError Handling

hashtagContext Manager Pattern

hashtagNotes

hashtagSee Also

Overview

Definition

Constructor Parameters

Namespaces

`.write` - Encrypted Write

`.read` - Encrypted Read

`.twrite` - Transparent Write

`.tread` - Transparent Read

`.dwrite` - Debug Write

Examples

Basic Encrypted Write

Encrypted Read

Transparent Operations

Debug Write

Concurrent Operations

With Transaction Parameters

Batch Processing

Using EIP-712 Signing

Instantiation via Async Client

Error Handling

Context Manager Pattern

Notes

See Also