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

> Easily test individual agent components and defined workflows with the Heurist Agent Frameworks built-in async test harness. Learn how to use the test_agent helper for efficient validation.

- Repository: [Heurist/heurist-agent-framework](https://github.com/heurist-network/heurist-agent-framework)
- Tags: testing
- Published: 2026-03-03

---

**The Heurist Agent Framework provides a built-in async test harness in [`mesh/tests/_test_agents.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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.

```python

# 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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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:

```bash
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`:

```python
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:

```python
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:

```python

# 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:

```bash

# 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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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`](https://github.com/heurist-network/heurist-agent-framework/blob/main/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.