Live Onchain Data in CrewAI: Build a Custom EVM Blockchain Tool

CrewAI ships tools for web search, file I/O, and code execution — but nothing for reading live EVM contract state. Ask a CrewAI agent for the current USDC balance of a wallet and it will either hallucinate a number from training data or refuse the task. That balance changes every block and was never in any corpus.

Fixing this takes one custom tool. Define a BaseTool subclass that POSTs to the evmquery REST API, assign it to whichever agent in the crew needs chain access, and any prompt about current token balances, oracle prices, or DeFi positions returns a live, decoded answer.

How CrewAI tools work

CrewAI tools extend BaseTool from crewai.tools. Each tool declares a name, a description the model reads when deciding whether to call it, and an args_schema — a Pydantic model that defines the input fields and their descriptions. The _run method contains the implementation.

When a CrewAI agent decides it needs external data, it emits a structured tool call matching the schema. The framework validates the arguments, routes the call to _run, and injects the return value back into the agent’s reasoning chain. You define the shape; the crew handles the routing.

For onchain queries this is the right design. The agent knows it cannot answer “what is the current Aave health factor?” from training data. A properly described tool gives it a deterministic path to the real answer without you hard-coding routing logic in application code.

CrewAI also supports a simpler @tool function decorator for quick one-offs. This post uses BaseTool because it provides a typed args_schema, which produces better model instructions and makes the tool easier to unit-test in isolation.

What you’ll build

A single EvmqueryReadTool that any agent in a crew can invoke. The tool accepts a chain, a named contract map, a CEL expression, and optional context variables. It calls the evmquery REST API and returns the decoded value and block number.

The same tool works in a single-agent workflow and in a multi-agent crew where, for example, a blockchain analyst agent fetches data and a portfolio reporter agent formats the findings for an end user.

If you want to query the chain interactively from Claude Desktop or Cursor without writing code, the evmquery MCP server is the faster path. This post is for production Python agents.

Setup

pip install crewai requests

Set two environment variables:

export ANTHROPIC_API_KEY=sk-ant-...
export EVMQUERY_API_KEY=evmq_...

Get a free evmquery key at https://app.evmquery.com/onboarding?plan=free. The free tier covers 2,000 credits per month — roughly 1,000 typical contract reads.

Defining the evmquery tool

import os
from typing import Optional, Type

import requests
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

EVMQUERY_API = "https://api.evmquery.com/api/v1/query"


class EvmqueryReadInput(BaseModel):
    chain: str = Field(
        description="Chain to query. One of: evm_ethereum, evm_base, evm_bnb_mainnet."
    )
    contracts: dict[str, str] = Field(
        description=(
            "Named contract addresses. Key is the short name used in the expression, "
            "value is the 0x contract address. "
            "Example: {'usdc': '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'}"
        )
    )
    expression: str = Field(
        description=(
            "CEL expression to evaluate. Named contracts become variables. "
            "Example: 'formatUnits(usdc.balanceOf(wallet), usdc.decimals())'"
        )
    )
    context: Optional[dict[str, str | list[str]]] = Field(
        default=None,
        description=(
            "Runtime values for wallet addresses or other parameters used in the expression. "
            "Pass a list of addresses for multi-wallet expressions. "
            "Example: {'wallet': '0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045'}"
        ),
    )


def _sel_type(v: str | list) -> str:
    return "list<sol_address>" if isinstance(v, list) else "sol_address"


class EvmqueryReadTool(BaseTool):
    name: str = "evmquery_read"
    description: str = (
        "Read live data from an EVM smart contract. Use for current token balances, "
        "DeFi positions, oracle prices, or any contract view function on Ethereum, "
        "Base, or BNB Smart Chain. Do NOT use for historical data or event logs."
    )
    args_schema: Type[BaseModel] = EvmqueryReadInput

    def _run(
        self,
        chain: str,
        contracts: dict[str, str],
        expression: str,
        context: Optional[dict] = None,
    ) -> str:
        body: dict = {
            "chain": chain,
            "schema": {
                "contracts": {k: {"address": v} for k, v in contracts.items()},
            },
            "expression": expression,
        }
        if context:
            body["schema"]["context"] = {k: _sel_type(v) for k, v in context.items()}
            body["context"] = context

        resp = requests.post(
            EVMQUERY_API,
            json=body,
            headers={"x-api-key": os.environ["EVMQUERY_API_KEY"]},
            timeout=10,
        )
        resp.raise_for_status()
        data = resp.json()
        result = data["result"]["value"]
        block = (data.get("meta") or {}).get("blockNumber", "?")
        return f"{result} (block {block})"

Three design choices worth noting:

• Field descriptions are the model’s schema. Pydantic Field(description=...) strings are what the model reads when deciding how to call the tool. Specific scope constraints — “Do NOT use for historical data” — reduce misrouted calls.
• Contracts expand to {"address": "..."} objects. The evmquery REST API expects contract entries as objects, not plain strings. The _run method handles this conversion so the agent passes simple name-to-address dicts and never sees the wire format.
• _sel_type infers context variable types. evmquery’s type system distinguishes single addresses (sol_address) from lists (list<sol_address>). The helper infers the correct declaration from the Python value — the model does not need to know about evmquery’s type system.

Three live recipes

All expressions below were validated against the live chain before publication.

ERC-20 balance check

tool = EvmqueryReadTool()

result = tool._run(
    chain="evm_ethereum",
    contracts={"usdc": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"},
    expression="formatUnits(usdc.balanceOf(wallet), usdc.decimals())",
    context={"wallet": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"},
)
# → "5567.402493 (block 24976057)"

formatUnits reads decimals() directly from the contract, so scaling is always correct regardless of whether the token uses 6, 8, or 18 decimal places.

Live ETH/USD price via Chainlink

result = tool._run(
    chain="evm_ethereum",
    contracts={"eth_usd": "0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419"},
    expression="formatUnits(eth_usd.latestAnswer(), eth_usd.decimals())",
)
# → "2287.04 (block 24976057)"

No context variables needed — this is a pure contract read with no wallet parameter. The address is the canonical Chainlink ETH/USD aggregator on Ethereum mainnet.

Aave v3 health factor

result = tool._run(
    chain="evm_ethereum",
    contracts={"aave": "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2"},
    expression="formatUnits(aave.getUserAccountData(wallet).healthFactor, 18)",
    context={"wallet": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"},
)
# → "1.157920892373162e+59 (block 24976059)"

A health factor above 1.0 means the position is safe; below 1.0 triggers liquidation. The large exponential result here represents a wallet with no active Aave debt — it is the maximum uint256 value scaled by 1e18, evmquery’s way of encoding infinite health. Add a note in your agent’s system prompt so it interprets this correctly rather than reporting an error.

Wiring into a CrewAI agent

With the tool class defined, assign it to an agent that needs chain access:

from crewai import Agent, Task, Crew, LLM

llm = LLM(model="claude-sonnet-4-6")

blockchain_analyst = Agent(
    role="Blockchain Data Analyst",
    goal=(
        "Fetch accurate, live onchain data to answer user questions about "
        "token balances and DeFi positions."
    ),
    backstory=(
        "You are a seasoned DeFi analyst with deep knowledge of EVM smart contracts. "
        "You always verify data by reading directly from the chain before drawing conclusions."
    ),
    tools=[EvmqueryReadTool()],
    llm=llm,
    verbose=True,
)

task = Task(
    description=(
        "Check the USDC balance of 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 on Ethereum "
        "and the current ETH/USD price from Chainlink. Report both values with block numbers."
    ),
    expected_output="A brief report with the USDC balance and ETH price, including block numbers.",
    agent=blockchain_analyst,
)

crew = Crew(agents=[blockchain_analyst], tasks=[task])
result = crew.kickoff()
print(result)

The agent sees evmquery_read in its toolbox, recognises the task requires live data, and calls the tool twice — once for the USDC balance and once for the ETH price. The block numbers in each result confirm the data is current, not cached.

Multi-agent workflow: analyst and reporter

CrewAI’s strength is composing multiple specialised agents. A blockchain analyst fetches raw data; a portfolio reporter formats it for a non-technical audience. Only the analyst gets the evmquery tool:

portfolio_reporter = Agent(
    role="Portfolio Reporter",
    goal="Turn raw blockchain data into clear, readable summaries for non-technical users.",
    backstory=(
        "You translate DeFi numbers into plain English. "
        "You never invent data — you wait for the analyst's findings before writing."
    ),
    llm=llm,
)

fetch_task = Task(
    description=(
        "Use evmquery_read to fetch the USDC balance and current ETH/USD price "
        "for 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 on Ethereum. "
        "Include the block number for each value."
    ),
    expected_output="Raw balance and price values with block numbers.",
    agent=blockchain_analyst,
)

report_task = Task(
    description=(
        "Using only the analyst's findings, write a two-sentence portfolio snapshot "
        "in plain English. Include the block numbers to show data freshness."
    ),
    expected_output="A clear, jargon-free portfolio snapshot with block numbers.",
    agent=portfolio_reporter,
    context=[fetch_task],
)

crew = Crew(
    agents=[blockchain_analyst, portfolio_reporter],
    tasks=[fetch_task, report_task],
)
result = crew.kickoff()
print(result)

context=[fetch_task] tells CrewAI that report_task depends on fetch_task’s output. The analyst runs first, fetches live data, and the result is passed as context to the reporter. The reporter never calls the chain directly — only the analyst has the tool.

Developers building DeFi-aware AI products

If you are building production DeFi tooling for AI agents, the developer resources page covers the full evmquery REST API: multi-wallet batch reads using the CEL map operator, list-type context variables for scanning many addresses in one call, and struct field access for protocols that return complex return types. For AI agent workflows specifically, the AI users page covers the REST tool and MCP surfaces side by side.

CrewAI tool vs MCP: picking the right surface

	CrewAI tool (this post)	MCP server
Language	Python	Any MCP client
Setup	~60 lines in your agent file	Paste one config block
Multi-agent	Yes, assign per-agent	Single conversation surface
Control	Full: schema, error handling, logging	Client manages the conversation
Best for	Production crews, custom backends	Claude Desktop, Cursor, VS Code

If you are building a Python multi-agent system, the BaseTool approach gives full control over the Pydantic schema, error formatting, and result shaping before the model sees the value. If you want to query the chain interactively from your IDE without writing code, the evmquery MCP server guide gets you there in under five minutes.

Next steps

• Set up the evmquery MCP server in Claude Desktop and Cursor, no code required
• Add a live EVM tool to LangChain agents in Python
• Monitor Aave health factors with a Python polling script
• Browse the evmquery REST API docs for multi-wallet macros, list filtering, and the full expression reference