Core API Reference

Complete API reference for the LeafMesh class.

LeafMesh Class

Constructors

`LeafMesh.from_store()`

Start the agency named by LEAFMESH_AGENCY_NAME from the store set in .env (2.4.481+). This is how new projects start. An empty store starts a blank agency, built in Studio.

from leafmesh import LeafMesh

leafmesh = LeafMesh.from_store()

From 2.4.481 new projects start with LeafMesh.from_store(). The agency's configuration lives in its store; from_store() starts the agency named by LEAFMESH_AGENCY_NAME from the store set in .env. An existing config.yaml is imported into the store once, then not read (the boot log says so).

`LeafMesh.from_yaml(yaml_path: str) -> LeafMesh`

Create a LeafMesh instance from a YAML configuration file.

from_yaml() also uses the store: the file is imported once, on the first start with an empty store, then ignored (with a warning if the file was edited afterwards). The file sits wherever your main.py points; the templates use configs/config.yaml.

from leafmesh import LeafMesh

leafmesh = LeafMesh.from_yaml("configs/config.yaml")

Parameters:

yaml_path (str): Path to YAML configuration file

Returns: Configured LeafMesh instance

Raises:

FileNotFoundError: Configuration file not found
ValidationError: Invalid YAML syntax or missing required fields

`LeafMesh.from_dict(config_dict: dict) -> LeafMesh`

Create a LeafMesh instance from a Python dictionary.

config = {
    "name": "my_system",
    "architecture": "managed_mesh",
    "agents": {
        "solver": {
            "name": "solver",
            "model": "gpt-4o-mini",
            "prompt": "You solve math problems.",
            "yields": {"answer": "number"}
        }
    }
}
leafmesh = LeafMesh.from_dict(config)

Parameters:

config_dict (dict): Configuration dictionary matching the YAML schema

Returns: Configured LeafMesh instance

Lifecycle Methods

`async start() -> None`

Start LeafMesh and initialize all components.

await leafmesh.start()

Connects to Redis, registers agents from YAML, starts the API server on port 18820, and begins processing requests.

Raises:

ConnectionError: Redis connection failed
RuntimeError: No LLM providers available (missing API keys)

`async stop() -> None`

Stop LeafMesh and clean up all resources.

await leafmesh.stop()

Gracefully drains in-flight requests (30s timeout), stops the API server, and disconnects from Redis.

Communication Methods

`async mesh_call(target, input_data, session_id=None) -> dict`

Process a request through the agent mesh. This is the primary entry point for all interactions.

result = await leafmesh.mesh_call(
    "support_request",                       # Entry point name from YAML
    {"user_message": "My API is down"},      # Input data
    session_id="session_001"                 # Optional session ID
)

Parameters:

target (str): Target entry point or agent name
input_data (dict): Input data for the agent
session_id (str, optional): Session identifier for conversation continuity

Returns: dict — The agent's structured output (matching its yields schema)

Intelligence Functions

Intelligence functions enhance or replace an agent's response. The function runs after the LLM call for LLM agents, or is the entire agent for programmatic agents. The function name must match the agent name in YAML — leafmesh.discover() finds them automatically.

async def solver(llm_response, input_data, context):
    """
    Args:
        llm_response (dict): Raw LLM response parsed against yields schema.
        input_data (dict):   The data passed to the agent.
        context (dict):      Request context (session_id, agent_name).

    Returns:
        dict: Must match the agent's yields schema.
    """
    return {
        "answer": llm_response.get("answer", 0),
        "steps": llm_response.get("steps", [])
    }

Pre-Compose Pipeline

`@pre_compose(...)`

Run deterministic Python before the LLM call to control what the LLM sees. Import pre_compose from leafmesh:

from leafmesh import pre_compose

@pre_compose(context_processor=lambda ctx: {"extra_context": "enriched data"})
async def analyzer(llm_response, input_data, context):
    return llm_response

Decorators

Decorator	Purpose
`@pre_compose(...)`	Pre-LLM data enrichment (import from `leafmesh`)
`leafmesh.chain(*functions)`	Chain functions into a pipeline
`leafmesh.compose(**processors)`	Composition decorator
`leafmesh.conditional_chain(condition, *functions)`	Conditional pipeline
`leafmesh.chain_with_results(*functions)`	Pipeline collecting all results

Auto-Discovery

`discover(path, pattern) -> int`

Auto-discover and register intelligence functions from files matching a pattern. Functions are matched to agents by name — the function name must match the agent name in YAML.

count = leafmesh.discover("./agency/", "*_agent.py")
print(f"Registered {count} intelligence functions")

Analytics Methods

Method	Returns
Studio's Dashboard	Aggregated usage statistics
Studio's Dashboard	Cache hit rates and savings
`leafmesh.get_status()`	System status dictionary
`leafmesh.list_agents()`	List of registered agent names

Evolution

Removed from the runtime SDK in v2.1.54. The in-process leafmesh.evolve_mesh_architecture() method now raises with a pointer to the standalone evolution service. Operators kick off runs from Studio — see Evolutionary Optimization.

Error Handling

from leafmesh import LeafMesh

try:
    result = await leafmesh.mesh_call(
        "solver",
        input_data={"message": "Hello"},
        session_id="session_1"
    )
except Exception as e:
    print(f"Error: {e}")

LeafMesh raises standard Python exceptions. Common error scenarios:

Agent not found: The target agent is not defined in YAML
Redis connection lost: Redis is unreachable
Mesh communication failed: Agent-to-agent routing error
Configuration invalid: YAML validation failed

Complete Example

import asyncio
from leafmesh import LeafMesh, pre_compose
from leafmesh.tools import global_tool

leafmesh = LeafMesh.from_store()

@global_tool(name="get_readings", description="Fetch sensor readings")
async def get_readings(sensor_id: str) -> dict:
    return {"readings": [95.2, 101.5, 87.3, 110.8]}

@pre_compose(
    input_processor=lambda input_data: {
        **input_data,
        "readings": [r for r in input_data.get("readings", []) if r is not None and -1000 < r < 10000]
    }
)
async def analyzer(llm_response, input_data, context):
    readings = input_data.get("readings", [])
    mean = sum(readings) / len(readings) if readings else 0
    return {
        "summary": llm_response.get("summary", ""),
        "mean": round(mean, 2),
        "count": len(readings),
        "severity": "high" if mean > 100 else "normal"
    }

async def main():
    await leafmesh.start()

    result = await leafmesh.mesh_call(
        "analyzer",
        input_data={"readings": [95.2, 101.5, 87.3, 110.8]},
        session_id="demo"
    )
    print(result)

    await leafmesh.stop()

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

Next Steps

Agent Configuration — Agent types and YAML schema
Configuration — Full YAML configuration reference
Tools — Tool system and @global_tool

LeafMesh — Configuration-first multi-agent orchestration