How to Test Individual Agent Components and Defined Workflows in Heurist Agent Framework

The Heurist Agent Framework provides a built-in async test harness in mesh/tests/_test_agents.py that lets you validate individual agent components and defined workflows by defining simple input dictionaries and invoking the test_agent helper.

Testing autonomous agents requires validating both isolated logic and end-to-end orchestration. The heurist-network/heurist-agent-framework provides a structured approach to test individual agent components and defined workflows through its Mesh testing infrastructure, allowing developers to verify tool schemas, API integrations, and reasoning patterns without deploying to production.

Architectural Overview of the Testing Infrastructure

The framework separates concerns across five core components to enable scalable agent testing.

Component	Role	Key Source
MeshAgent	Base class for every Mesh agent implementing the common lifecycle (__aenter__, __aexit__), tool registration (get_tool_schemas), and the entry point handle_message used by the test harness.	mesh/mesh_agent.py
Test Harness	Reusable async utilities that instantiate an agent, drive test cases, capture timings, and optionally write YAML reports.	mesh/tests/_test_agents.py
Individual Test Scripts	Self-contained modules that import an agent, declare a dictionary of test cases, and invoke test_agent.	mesh/test_scripts/test_token_resolver.py
Parallel Runner	Discovers every *.py file in mesh/test_scripts/ and runs them concurrently with default 5 workers.	mesh/tests/run_tests.py
Workflow Modules	Higher-level reasoning patterns (e.g., ResearchWorkflow) implemented as regular Python classes and invoked from an agent’s tool logic.	mesh/workflows/

How the Pieces Fit Together

1. Agent code in mesh/agents/<agent>.py extends MeshAgent.
2. The test harness in mesh/tests/_test_agents.py creates an instance, configures quiet logging, and calls await agent.handle_message(test_input).
3. Each test script supplies a dictionary of named cases (query, tool arguments, expected behavior). The harness iterates, respects optional delay_seconds to avoid rate-limit throttling, and records success or failure plus elapsed time.
4. run_tests.py launches every script in its own subprocess, aggregates logs, and prints a final summary.

Because the harness works at the agent level via handle_message, it automatically validates all internal workflow steps—tool schema validation, external API calls wrapped with @with_retry/@with_cache, post-processing, and final response formatting.

How to Test Individual Agent Components

Writing a Minimal Test Script

Create a new file in mesh/test_scripts/ following this template to validate your agent’s tool logic.

# mesh/test_scripts/test_my_custom_agent.py

import asyncio
from pathlib import Path
import sys

# Make repository root importable

sys.path.append(str(Path(__file__).parents[2]))

from mesh.agents.my_custom_agent import MyCustomAgent
from mesh.tests._test_agents import test_agent

# ------------------------------------------------------------------

# Define test cases – each key is a descriptive name.

# ------------------------------------------------------------------

TEST_CASES = {
    "basic_query": {
        "input": {"tool": "search", "tool_arguments": {"query": "BTC"}},
        "description": "Simple token search by symbol",
    },
    "profile_with_pairs": {
        "input": {
            "tool": "profile",
            "tool_arguments": {"symbol": "ETH", "include": ["pairs"]},
        },
        "description": "Retrieve ETH profile together with top trading pairs",
    },
}

# ------------------------------------------------------------------

# Execute the tests. 1‑second delay avoids hitting rate limits.

# ------------------------------------------------------------------

if __name__ == "__main__":
    asyncio.run(test_agent(MyCustomAgent, TEST_CASES, delay_seconds=1))

Key implementation details:

• The test_agent helper is imported from mesh/tests/_test_agents.py.
• Each test case maps a name to an input dictionary and a description.
• The delay_seconds parameter prevents HTTP 429 errors when calling external APIs like CoinGecko or DexScreener.

Configuring Test Parameters

Control verbosity and timing through environment variables and function arguments.

Adjust log verbosity using MESH_TEST_LOG_LEVEL when debugging failures:

export MESH_TEST_LOG_LEVEL=INFO
uv run python mesh/test_scripts/my_test.py

Avoid rate-limit throttling by passing delay_seconds to test_agent:

await test_agent(MyCustomAgent, TEST_CASES, delay_seconds=1.5)

Generate machine-readable reports by inspecting the YAML output written alongside your test script. The harness automatically exports *_example.yaml files containing timings and results for downstream dashboards.

How to Test Defined Workflows

Integration Testing via Agents

Workflows such as ResearchWorkflow or ChainOfThoughtReasoning reside in mesh/workflows/ and are invoked from an agent’s tool logic. Because the test harness calls handle_message, it exercises the complete workflow chain—including tool schema validation, external API calls decorated with @with_retry and @with_cache, and final response formatting.

To validate a workflow, simply write a test case that triggers the agent tool using it:

TEST_CASES = {
    "research_workflow": {
        "input": {
            "tool": "research",
            "tool_arguments": {"topic": "decentralized AI"}
        },
        "description": "Validates ResearchWorkflow end-to-end",
    }
}

Direct Workflow Unit Testing

For isolated validation of workflow logic without agent overhead, instantiate the workflow class directly in a pytest-style test:

# mesh/tests/test_research_workflow.py

import pytest
from mesh.workflows.research import ResearchWorkflow

@pytest.mark.asyncio
async def test_simple_research():
    wf = ResearchWorkflow()
    result = await wf.run(query="Heurist AI")
    assert "summary" in result
    assert isinstance(result["sources"], list)

This pattern is useful when developing new reasoning patterns in mesh/workflows/ before integrating them into an agent.

Running the Full Test Suite

Execute every test script concurrently using the parallel runner for CI-style validation:

# From the repository root

uv sync                # install deps (once)

uv run python mesh/tests/run_tests.py   # executes every test script concurrently

The runner in mesh/tests/run_tests.py discovers all *.py files in mesh/test_scripts/, launches them with default 5 workers, aggregates logs, and prints a final summary. This ensures that changes to shared base classes or workflow utilities do not break individual agent implementations.

Summary

• Use the built-in harness: Import test_agent from mesh/tests/_test_agents.py to avoid boilerplate when validating agent logic.
• Test at the agent level: The handle_message entry point automatically exercises tool schemas, external API calls with retry/cache decorators, and internal workflow steps.
• Control execution: Set MESH_TEST_LOG_LEVEL for debugging verbosity and use delay_seconds to prevent rate-limit errors during external API calls.
• Scale with parallel runners: Use mesh/tests/run_tests.py to execute the entire suite concurrently for regression testing.
• Isolate workflow logic: For complex reasoning patterns in mesh/workflows/, write direct pytest-style unit tests in addition to integration tests through agents.

Frequently Asked Questions

What is the fastest way to test a single agent during development?

Create a minimal test script in mesh/test_scripts/ that imports your agent class and the test_agent helper from mesh/tests/_test_agents.py. Define a dictionary with one or two test cases mapping input tools to expected behaviors, then call asyncio.run(test_agent(YourAgent, TEST_CASES)). This avoids the overhead of the parallel runner and gives immediate feedback on tool logic.

How do I handle rate limits when testing agents that call external APIs?

Pass the delay_seconds parameter to test_agent to introduce a sleep interval between test cases. For example, await test_agent(MyAgent, TEST_CASES, delay_seconds=1.5) waits 1.5 seconds between executions, preventing HTTP 429 errors from services like CoinGecko or DexScreener. You can also set MESH_TEST_LOG_LEVEL=INFO to see exact request timings.

Can I test workflow logic without running the full agent?

Yes. Workflow classes in mesh/workflows/ are standard Python classes that can be instantiated and exercised directly. Write a pytest-style test that imports the workflow—for example from mesh.workflows.research import ResearchWorkflow—and call its run() method with test inputs. This isolates reasoning logic from agent lifecycle management and external tool calls.

How do I run all agent tests in parallel for CI validation?

Execute uv run python mesh/tests/run_tests.py from the repository root. This script discovers every *.py file in mesh/test_scripts/, runs them concurrently with five workers by default, aggregates stdout/stderr, and prints a final summary. It is the recommended entry point for continuous integration pipelines to catch regressions across the entire agent mesh.

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