Blockchain Monitoring in Python: Poll EVM Contract State Without the Boilerplate

Most Python blockchain monitoring scripts start the same way. You want one number, typically a wallet balance, a health factor, or a pool reserve, on a schedule, with an alert if it crosses a threshold. You end up with 80 lines of RPC plumbing before you touch the alert logic.

This guide uses evmquery’s REST API instead. Send a named contract address and a CEL expression, get the decoded result back. The multi-wallet .map() and .filter() macros batch internally. A working Aave health-factor monitor that checks a wallet list every minute comes out to roughly 45 lines of Python.

Why raw RPC breaks down for polling

The standard approach for reading contract state from Python is web3.py against a JSON-RPC endpoint. That works for one-off scripts. It fights you when you want a production monitor:

• Multiple wallets, one call. Multicall3 batching means constructing calldata, encoding ABI selectors, bundling into the aggregate3 struct, and decoding bytes responses. Each contract type needs its own encoder/decoder pair.
• Proxy contracts. Aave Pool, Uniswap v4’s pool manager, most of DeFi sit behind EIP-1967 proxies. Your ABI needs to be the implementation’s, which means a second eth_call to implementation() before you can start.
• Typed results. web3.py decodes a (uint256, uint256, uint256, uint256, uint256, int256) tuple correctly, but you still need to know which slot is healthFactor, then scale it from 1e18 fixed-point.

For a monitor running every 60 seconds against 10 wallets and two chains, “solvable but tedious” compounds fast. The boilerplate dwarfs the logic.

The evmquery REST API

evmquery exposes a single endpoint for contract reads:

POST https://api.evmquery.com/api/v1/query
x-api-key: <your-key>

The JSON body has three fields:

Field	Type	Purpose
chain	string	evm_ethereum, evm_base, or evm_bnb_mainnet
schema	object	contracts (name → address map) + optional context (typed variable declarations)
expression	string	A CEL expression over the named contracts and context variables

The optional top-level context field pairs with schema.context: declare types there, pass runtime values here. That’s the whole API surface.

Grab an API key from the evmquery dashboard; the free tier gives you 2,000 credits/month.

Your first contract read

pip install httpx

A thin wrapper around the endpoint, then a USDC balance check:

import os
import httpx

API_KEY = os.environ["EVMQUERY_API_KEY"]
ENDPOINT = "https://api.evmquery.com/api/v1/query"


def execute(
    chain: str,
    contracts: dict,
    expression: str,
    context_types: dict | None = None,
    context_values: dict | None = None,
):
    schema = {"contracts": contracts}
    if context_types:
        schema["context"] = context_types
    body = {"chain": chain, "schema": schema, "expression": expression}
    if context_values:
        body["context"] = context_values
    resp = httpx.post(ENDPOINT, headers={"x-api-key": API_KEY}, json=body, timeout=10)
    resp.raise_for_status()
    return resp.json()["result"]


# Single-wallet USDC balance
balance = execute(
    chain="evm_ethereum",
    contracts={"usdc": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"},
    expression="formatUnits(usdc.balanceOf(wallet), usdc.decimals())",
    context_types={"wallet": "sol_address"},
    context_values={"wallet": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"},
)
print(balance)  # → 5567.402493

formatUnits scales the raw uint256 by the token’s decimals. usdc.decimals() is a second on-chain read; both calls are batched automatically. The result is a plain Python float.

Multi-wallet balance checks with .map()

The CEL .map() macro applies an expression to every element of a list. evmquery batches all the resulting contract calls via Multicall3 internally, so checking 20 wallets is one HTTP request, not 20.

WALLETS = [
    "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    "0x95222290DD7278Aa3Ddd389Cc1E1d165CC4BAfe5",
    "0x4838B106FCe9647Bdf1E7877BF73cE8B0BAD5f97",
]

balances = execute(
    chain="evm_ethereum",
    contracts={"usdc": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"},
    expression="wallets.map(w, formatUnits(usdc.balanceOf(w), usdc.decimals()))",
    context_types={"wallets": "list<sol_address>"},
    context_values={"wallets": WALLETS},
)
# → [5567.402493, 1510.424957, 343.649519]

The result list aligns with the input list: balances[i] is the balance for WALLETS[i]. If you’ve built a Multicall3 wrapper before, this is what it looks like once the plumbing is gone.

Aave health factor alerts with .filter()

Health factor below 1.0 means liquidation. Below 1.2 means you’re close. .filter() returns only the list elements that match a predicate, so you get back only the wallets that need attention.

AAVE_POOL = "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2"


def wallets_at_risk(wallets: list[str], threshold: float = 1.2) -> list[str]:
    """Return addresses whose Aave v3 health factor is below `threshold`."""
    threshold_scaled = str(int(threshold * 10**18))
    return execute(
        chain="evm_ethereum",
        contracts={"aave": AAVE_POOL},
        expression=f"wallets.filter(w, aave.getUserAccountData(w).healthFactor < parseUnits('{threshold}', 18))",
        context_types={"wallets": "list<sol_address>"},
        context_values={"wallets": wallets},
    )

getUserAccountData returns a struct with six fields. The expression drills into .healthFactor directly; evmquery resolves the proxy to the Aave Pool implementation and decodes the struct automatically.

Here is the complete monitor loop:

import time


def send_alert(wallets: list[str]) -> None:
    import httpx
    httpx.post(
        os.environ["SLACK_WEBHOOK"],
        json={"text": f":warning: Aave health factor < 1.2 on Ethereum:\n" + "\n".join(wallets)},
    )


def monitor(wallets: list[str], interval_seconds: int = 60) -> None:
    print(f"Monitoring {len(wallets)} wallet(s) every {interval_seconds}s…")
    while True:
        at_risk = wallets_at_risk(wallets)
        if at_risk:
            send_alert(at_risk)
            print(f"Alert sent for {len(at_risk)} wallet(s).")
        else:
            print("All positions safe.")
        time.sleep(interval_seconds)


if __name__ == "__main__":
    monitor(WALLETS)

One request per interval, one decoded list, conditional alert. That is the entire monitoring script.

Cross-chain checks on Ethereum, Base, and BNB Smart Chain

evmquery supports Ethereum, Base, and BNB Smart Chain. A user who runs positions on multiple networks needs two or three calls, not one per chain (cross-chain expressions in a single call are not supported yet; each call targets one chain).

AAVE_POOLS = {
    "evm_ethereum": "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2",
    "evm_base":     "0xA238Dd80C259a72e81d7e4664a9801593F98d1c5",
}


def health_factor(wallet: str, chain: str, pool_address: str) -> float:
    return float(execute(
        chain=chain,
        contracts={"aave": pool_address},
        expression="formatUnits(aave.getUserAccountData(wallet).healthFactor, 18)",
        context_types={"wallet": "sol_address"},
        context_values={"wallet": wallet},
    ))


wallet = "0xYourWallet"
for chain, pool in AAVE_POOLS.items():
    hf = health_factor(wallet, chain, pool)
    label = chain.replace("evm_", "")
    print(f"{label}: {hf:.4f}")

Two sequential calls, both under 50ms each typically. Fire them concurrently with asyncio.gather if you’re polling many wallets across all chains and latency matters.

If you’re building this for automation workflows, the same expressions that run in Python run in the n8n community node, so a working Python prototype translates directly to a no-code workflow.

Scheduling the monitor

Cron (simplest, on any Linux host):

# Run every 5 minutes; API key in environment
*/5 * * * * EVMQUERY_API_KEY=evq_... SLACK_WEBHOOK=https://... python3 /opt/monitor.py

GitHub Actions (no server required):

name: Aave Health Monitor
on:
  schedule:
    - cron: "*/15 * * * *"  # every 15 minutes
jobs:
  monitor:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install httpx && python monitor.py
        env:
          EVMQUERY_API_KEY: ${{ secrets.EVMQUERY_API_KEY }}
          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}

GitHub Actions cron has a minimum granularity of one minute and runs for free within the included quota. The secrets store keeps the API key out of the repository.

From Python script to REST call

The full flow for any new contract is the same three steps:

1. Name the contract in schema.contracts, using a short key.
2. Write the expression: call methods by name, chain helpers like formatUnits, use .map() or .filter() for lists.
3. Declare context variables in schema.context if the expression is parameterized, and pass values in the top-level context field.

For unfamiliar contracts, the describe_schema tool in the evmquery MCP server lists every callable view/pure method with parameter types. Run it in Claude Code before you write the expression and you’ll know exactly what’s available.

Next steps

• Want the same expressions in a no-code workflow? Read smart contracts in n8n covers the community node and three paste-in recipes.
• Building for an AI assistant instead of a script? The MCP server guide wires evmquery directly into Claude, Cursor, and VS Code.
• Comparing this approach to raw Alchemy or QuickNode RPC? The Moralis / Alchemy / QuickNode comparison maps out which layer fits which use case.
• More details on the query engine and supported chains: evmquery for developers.