How to Integrate a New Heurist Mesh Agent to Leverage Web3 Data

Integrating a new Heurist Mesh agent requires creating a Python class that inherits from MeshAgent, placing it in the mesh/agents/ directory, and implementing three core methods—get_system_prompt, get_tool_schemas, and _handle_tool_logic—so the AgentLoader automatically discovers and registers your agent for LLM-driven Web3 queries.

The heurist-network/heurist-agent-framework provides a modular architecture for building AI agents that interact with blockchain data. When you integrate a new Heurist Mesh agent to leverage Web3 data, you create a specialized module that auto-registers with the Mesh ecosystem, enabling instant access via the Python client library or HTTP API for on-chain analytics and DeFi operations.

Step-by-Step Implementation Guide

Create the Agent File in mesh/agents/

All Mesh agents reside in the mesh/agents/ package. Create a new Python file, such as my_web3_agent.py, alongside existing implementations like evm_token_info_agent.py. No manual registration is required—the AgentLoader in mesh/mesh_manager.py scans this directory using pkgutil.iter_modules (lines 31-78) and instantiates every concrete MeshAgent subclass automatically.

Subclass MeshAgent and Configure Metadata

Inherit from the abstract base class located in mesh/mesh_agent.py and update the metadata dictionary to enable discovery, pricing, and UI rendering.

# mesh/agents/my_web3_agent.py

import os
import logging
from typing import Any, Dict, List, Optional
from dotenv import load_dotenv
from decorators import monitor_execution, with_cache, with_retry
from mesh.mesh_agent import MeshAgent

logger = logging.getLogger(__name__)
load_dotenv()

class MyWeb3Agent(MeshAgent):
    def __init__(self):
        super().__init__()
        self.api_key = os.getenv("MY_WEB3_API_KEY")
        if not self.api_key:
            raise ValueError("MY_WEB3_API_KEY environment variable is required")
        self.headers = {"Authorization": f"Bearer {self.api_key}"}
        self.base_url = "https://api.myweb3provider.io/v1"
        
        # Metadata drives discovery and pricing

        self.metadata.update({
            "name": "My Web3 Analytics Agent",
            "version": "1.0.0",
            "author": "Your Name",
            "author_address": "0xYOURADDRESS",
            "description": "Fetches on-chain token metrics from MyWeb3Provider.",
            "tags": ["EVM", "DeFi"],
            "image_url": "https://example.com/my_web3_agent.png",
            "x402_config": {"enabled": True, "default_price_usd": "0.005"},
        })

Define System Prompt and Tool Schemas

Implement get_system_prompt to instruct the LLM on your agent's domain, and get_tool_schemas to declare JSON Schema-compatible function signatures for Gemini or ChatGPT function calling.

    def get_system_prompt(self) -> str:
        return (
            "You are a DeFi data extractor that can query MyWeb3Provider for token "
            "price, holder distribution and recent large trades. Return raw data "
            "or a concise explanation when asked."
        )

    def get_tool_schemas(self) -> List[Dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": "get_token_stats",
                    "description": "Retrieve price, market-cap and holder count for an ERC-20 token.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "chain": {
                                "type": "string",
                                "description": "Blockchain network (e.g. ethereum, bsc, avalanche)",
                            },
                            "address": {
                                "type": "string",
                                "description": "ERC-20 contract address, must start with 0x",
                            },
                        },
                        "required": ["chain", "address"],
                    },
                },
            }
        ]

Implement Tool Logic with Web3 Data Sources

Write the async method that queries your Web3 data provider. Use self._api_request (inherited from MeshAgent) to handle HTTP GET/POST, automatic timeout handling, optional proxy fallback, and caching. Apply the @with_cache and @with_retry decorators from decorators.py for resilience.

    @with_cache(ttl_seconds=300)
    @with_retry(max_retries=2)
    async def get_token_stats(self, chain: str, address: str) -> Dict[str, Any]:
        if not address.startswith("0x"):
            return {"error": "Invalid contract address"}
        endpoint = f"{self.base_url}/token/{chain}/{address}"
        return await self._api_request(
            url=endpoint, method="GET", headers=self.headers
        )

Handle Tool Dispatching

Override _handle_tool_logic to route incoming tool calls to your implementation. This method receives the tool_name, function_args, and optional session_context, returning a dict that feeds back into the LLM or directly to the caller.

    async def _handle_tool_logic(
        self,
        tool_name: str,
        function_args: dict,
        session_context: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if tool_name != "get_token_stats":
            return {"error": f"Unsupported tool {tool_name}"}
        return await self.get_token_stats(
            chain=function_args.get("chain"),
            address=function_args.get("address"),
        )

Configure Timeouts and Fallbacks (Optional)

Override get_default_timeout_seconds or get_tool_timeout_seconds if your Web3 endpoint requires stricter limits. Implement get_fallback_for_tool to specify backup agent logic when timeouts occur.

    def get_default_timeout_seconds(self) -> Optional[int]:
        return 8

Auto-Discovery and Registration Mechanics

The framework uses dynamic discovery—no explicit registration code is necessary. When the service starts, AgentLoader.load_agents() builds a dictionary mapping {AgentClassName: AgentClass}. This registry is consumed by main_api.py for the HTTP server and by heurist_mesh_client/client.py for the Python SDK. As long as your file is importable (the __init__.py in mesh/agents/ exists), your agent becomes available instantly.

Testing Your New Agent Locally

Local Test Script

Create a script under mesh/test_scripts/ to verify your agent before deployment. Use the call_agent method to simulate incoming requests.

# mesh/test_scripts/run_my_web3_agent.py

import asyncio
import sys
from pathlib import Path

project_root = Path(__file__).parents[2]
sys.path.insert(0, str(project_root))

from mesh.agents.my_web3_agent import MyWeb3Agent

async def main():
    agent = MyWeb3Agent()
    payload = {
        "tool": "get_token_stats",
        "tool_arguments": {
            "chain": "ethereum",
            "address": "0x6B175474E89094C44Da98b954EedeAC495271d0F",  # DAI

        },
        "raw_data_only": True,
    }
    result = await agent.call_agent(payload)
    print("Result:", result)

if __name__ == "__main__":
    asyncio.run(main())

Run the test from the repository root:

uv run python mesh/test_scripts/run_my_web3_agent.py

Using the Mesh Client Library

Once registered, invoke your agent through the official client library demonstrated in examples/basic_usage.py:

from heurist_mesh_client.client import HeuristMeshClient

async def demo():
    client = HeuristMeshClient()
    response = await client.call_agent(
        agent_name="MyWeb3Agent",
        params={
            "tool": "get_token_stats",
            "tool_arguments": {
                "chain": "bsc", 
                "address": "0xe9e7cea3dedca5984780bafc599bd69add087d56"
            },
        },
    )
    print(response)

Integration with the Mesh Ecosystem

LLM-driven routing: When users submit free-form queries, the base handle_message method (in MeshAgent) forwards requests to Gemini (mesh/gemini.py). Gemini selects the appropriate tool based on your declared schemas, then _execute_tool_with_policy manages execution with timeout and fallback logic.

Micro-payments: If you configure x402_config in your agent's metadata, the Mesh runtime automatically charges callers via the x402 micro-payment layer (see mesh/erc8004/ for on-chain registration details).

Environment variables: Add required keys (e.g., MY_WEB3_API_KEY) to .env.example so other developers know which credentials to configure.

Summary

• Place new agent files in mesh/agents/ to trigger auto-discovery by AgentLoader in mesh/mesh_manager.py.
• Inherit from MeshAgent and implement get_system_prompt, get_tool_schemas, and _handle_tool_logic to define capabilities.
• Use self._api_request for HTTP calls to Web3 data providers, leveraging built-in timeout, proxy, and caching support.
• Decorate async tool methods with @with_cache and @with_retry from decorators.py for production resilience.
• Test locally using mesh/test_scripts/ or the HeuristMeshClient before deploying to the Mesh API.

Frequently Asked Questions

Do I need to manually register my agent with the Mesh Manager?

No. The AgentLoader class in mesh/mesh_manager.py automatically discovers any concrete subclass of MeshAgent located in the mesh/agents/ package using pkgutil.iter_modules. Simply placing your Python file in this directory with a valid class definition makes it available to the API and client library immediately upon service restart.

How does the framework handle Web3 API authentication?

Store sensitive credentials in environment variables (e.g., BITQUERY_API_KEY or MY_WEB3_API_KEY), load them in your agent's __init__ method, and pass them via headers in self._api_request. Include these variable names in .env.example so users know what to configure, but never hardcode keys in your agent's source code.

Can I charge users for accessing my Web3 agent?

Yes. Include an x402_config dictionary in your self.metadata.update() call with "enabled": True and a "default_price_usd" value. The Mesh runtime inspects this configuration and automatically processes micro-payments through the x402 layer before executing your tool logic.

What happens if my Web3 data provider is slow or unavailable?

The MeshAgent base class provides timeout handling through get_default_timeout_seconds and get_tool_timeout_seconds. If a call exceeds these limits, you can implement get_fallback_for_tool to return data from a backup agent or cached response. Additionally, the @with_retry decorator automatically retries transient failures up to your specified max_retries.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}