Agent Development

Practical guide for building, testing, and organizing agents in LeafMesh.

Development Workflow

Define agents in YAML — prompts, yields, can_call rules
Write intelligence functions — deterministic Python logic
Test locally — run against a local Redis instance
Iterate — adjust prompts, conditions, and logic

File Organization

For systems with more than 3 agents, organize into separate files:

my_system/
├── .env                     # Agency name and store (configuration lives in the store)
├── main.py                  # Entry point
├── agency/
│   ├── __init__.py
│   ├── agency_client.py     # LeafMesh init + agent registration
│   ├── solver_agent.py      # Solver intelligence
│   ├── checker_agent.py     # Checker intelligence
│   └── tools/
│       └── database.py      # Custom tool definitions

Agent Files

# agency/solver_agent.py
# Function name "solver" matches the YAML agent name — auto_discover finds it
async def solver(llm_response, input_data, context):
    return llm_response

# agency/agency_client.py
from leafmesh import LeafMesh

async def initialize():
    leafmesh = LeafMesh.from_store()
    await leafmesh.start()
    # auto_discover in YAML config handles agent registration
    # Set auto_discover: true and agents_dir in the agency's configuration
    return leafmesh

Writing Intelligence Functions

Function Signature

# Function name matches the YAML agent name — auto_discover finds it
async def agent_name(llm_response: dict, input_data: dict, context: dict) -> dict:
    """
    llm_response: Parsed LLM output (empty dict for programmatic agents)
    input_data:   Data from caller or upstream agent's yields
    context:      {"session_id": "...", "agent_name": "..."}

    Returns: dict matching the agent's yields schema
    """
    return llm_response

Pattern: Deterministic Override

Replace the LLM response for cases where you can compute the answer:

# Function name "solver" matches the YAML agent name — auto_discover finds it
async def solver(llm_response, input_data, context):
    import re
    problem = input_data.get("problem", "")
    numbers = re.findall(r'-?\d+\.?\d*', problem)

    if len(numbers) == 2 and "+" in problem:
        a, b = float(numbers[0]), float(numbers[1])
        return {"answer": a + b, "confidence": 1.0}

    return llm_response  # Fall back to LLM

Pattern: Enrichment

Add computed fields to the LLM response:

# Function name "reporter" matches the YAML agent name — auto_discover finds it
async def reporter(llm_response, input_data, context):
    record_count = input_data.get("record_count", 0)
    error_count = len(input_data.get("errors", []))
    grade = "A" if error_count == 0 else "B" if error_count < 3 else "C"

    return {
        "summary": llm_response.get("summary", ""),
        "quality_grade": grade
    }

Pattern: Stateful Tracking

Track state across calls using yields and session context:

# Function name "stateful_agent" matches the YAML agent name — auto_discover finds it
async def stateful_agent(llm_response, input_data, context):
    # Use input_data from upstream agents and context for session awareness
    previous_count = input_data.get("call_count", 0)

    return {**llm_response, "call_count": previous_count + 1}

State persistence (sessions, yields, agent data) is handled automatically by the mesh and stored in Redis. You can query state via the REST API at :18820.

Testing

Unit Testing Intelligence Functions

Intelligence functions are regular async Python functions. Test them directly:

import pytest

@pytest.mark.asyncio
async def test_solver_addition():
    result = await solver_logic(
        llm_response={"answer": 5},
        input_data={"problem": "2 + 3"},
        context={"session_id": "test"}
    )
    assert result["answer"] == 5.0
    assert result["confidence"] == 1.0


@pytest.mark.asyncio
async def test_solver_fallback():
    """Complex problems fall back to LLM response"""
    llm_response = {"answer": 42, "confidence": 0.8}
    result = await solver_logic(
        llm_response=llm_response,
        input_data={"problem": "What is the meaning of life?"},
        context={"session_id": "test"}
    )
    assert result == llm_response

Integration Testing

@pytest.mark.asyncio
async def test_full_pipeline():
    leafmesh = LeafMesh.from_store()   # the test agency, named in the test .env
    await leafmesh.start()

    try:
        result = await leafmesh.mesh_call(
            "solver",
            input_data={"problem": "2 + 1"},
            session_id="integration_test"
        )
        assert "answer" in result
    finally:
        await leafmesh.stop()

Debugging

Inspect Session State

Browse session state, agent yields, and mesh communications in Studio's Sessions and Activity tabs. For aggregated runtime statistics from your own code, use Studio's Dashboard.

Common Issues

Problem	Cause	Fix
Agent not found	Function name doesn't match YAML agent name	Ensure function name matches agent name exactly
Empty response	Intelligence function not found by auto_discover	Ensure function is in the agents_dir and name matches YAML
can_call not triggering	Condition doesn't match yields	Check yield field names and types
Timeout errors	LLM provider slow or unreachable	Check API keys, network connectivity

Next Steps

Agent Types — All four agent types in detail
Testing — Testing framework guide
Debugging — Debugging tools and techniques

LeafMesh — Build, test, iterate