LeafMesh Class

The main entry point for LeafMesh. Manages the complete lifecycle of a multi-agent system.

Import

from leafmesh import LeafMesh

Construction

`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.

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(path: str) -> LeafMesh`

Load configuration from a YAML file:

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

Validates: file existence, size (< 10MB), YAML syntax, required fields, Pydantic models, UTF-8 encoding.

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.

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

leafmesh = LeafMesh.from_dict({
    "name": "my_system",
    "architecture": "managed_mesh",
    "agents": { ... }
})

`LeafMesh(config: Optional[LeafMeshConfig] = None)`

Direct construction. If no config provided, creates a default LeafMeshConfig().

Context Manager

async with LeafMesh.from_store() as leafmesh:
    result = await leafmesh.mesh_call("support_request", data, session_id="s1")

Lifecycle Methods

`await leafmesh.start()`

Initializes all internal components, connects to Redis, registers agents from YAML configuration, starts the API server on port 18820, and begins processing requests.

await leafmesh.start()

`await leafmesh.stop()`

Graceful shutdown. Drains in-flight requests (30s timeout), stops the API server, and disconnects from Redis.

await leafmesh.stop()

Request Processing

`await leafmesh.mesh_call(entry_point, data, session_id=None, *, wait_for_result=False, timeout=60, stream=False) -> Any`

Route a request through a configured entry point. Three return modes — pick the one that matches your use case:

# 1. Fire-and-forget (DEFAULT) — returns immediately with a dispatch ack.
#    The agent's actual output flows through normal mesh routing
#    (can_call / fan-in / chain). Use when you don't need a synchronous
#    response — e.g. webhook handlers, background triggers, agents
#    that fan out to multiple downstreams.
ack = await leafmesh.mesh_call(
    "support_request",
    {"user_message": "My API is down"},
    session_id="session_001",
)
# ack → {"status": "entry_point_event_published",
#         "entry_point": "support_request",
#         "target_agent": "greeter_agent",
#         "session_id": "session_001",
#         "processing": "pure_event_driven_asynchronous"}

# 2. Wait for the agent's actual output — blocks up to `timeout` seconds.
#    Use when you need the synchronous return value (e.g. a CLI tool, a
#    REST handler responding to the caller).
result = await leafmesh.mesh_call(
    "support_request",
    {"user_message": "My API is down"},
    session_id="session_001",
    wait_for_result=True,
    timeout=60,
)
# result → the agent's actual output dict

# 3. Stream tokens as they're generated (v2.2.1+). For LLM free-text agents,
#    real token streaming; for yields agents, a single final chunk.
async for chunk in await leafmesh.mesh_call(
    "support_request",
    data,
    session_id="session_001",
    stream=True,
):
    if chunk["type"] == "token":
        print(chunk["delta"], end="", flush=True)
    elif chunk["type"] == "final":
        final = chunk["data"]

Argument	Type	Default	Description
`entry_point`	string	—	Entry point name from YAML `entry_points:` block
`data`	any	—	Input data passed to the target agent
`session_id`	string	`None`	Session this call belongs to; auto-generated UUID if omitted
`wait_for_result`	bool	`False`	If `True`, block until the agent produces output and return it. If `False`, return a dispatch ack immediately.
`timeout`	int	`60`	Max seconds to wait when `wait_for_result=True`. After timeout, returns a timeout error.
`stream`	bool	`False`	Return an async iterator of chunks (mutually exclusive with `wait_for_result`).

This is the primary way to send requests into the system. The entry point maps to a target agent, which triggers the full pipeline: pre-compose → prompt assembly → LLM call → tool execution → yields parsing → can_call evaluation → intelligence function → downstream mesh routing.

Same defaults apply to the HTTP equivalent POST /api/mesh/request — fire-and-forget unless you opt in to ?wait_for_result=true.

`await leafmesh.rerun_agent(agent_name, session_id, *, feedback=None, reason="user_request", new_input=None) -> dict`

Re-run a single agent inside an existing session, optionally with feedback so it can self-correct (leafmesh >= 1.0.299). Use this to:

Wire a "Rerun" button in your own UI
Build custom retry rules outside Manager's analysis-driven path
Trigger a deterministic rerun from a debugging script

Routes through the same Manager retry path used by strict yields enforcement and analysis-driven retries. See Manager — Retry & Rerun for the full flow.

# Simple retrigger — same input as last time, no feedback
result = await leafmesh.rerun_agent(
    agent_name="advisor_agent",
    session_id="session-1777755-wrh93qrf",
)

# Retrigger with feedback (e.g. the caller spotted a bad shape)
result = await leafmesh.rerun_agent(
    agent_name="advisor_agent",
    session_id="session-1777755-wrh93qrf",
    feedback={
        "previous_output": {"recommendation": "vague answer"},
        "error": "Recommendation lacks specific action items",
        "expected_shape": {"recommendation": "string", "action_items": "list"},
    },
    reason="schema_mismatch",
)

# Retrigger with brand-new input (deliberately steer the agent)
result = await leafmesh.rerun_agent(
    agent_name="processor_agent",
    session_id="session-1777755-wrh93qrf",
    new_input={"message": "Now check my refund instead", "request_type": "refund"},
)

Argument	Type	Default	Description
`agent_name`	string	—	Agent to re-run (must be registered in this mesh).
`session_id`	string	—	Session this rerun belongs to.
`feedback`	dict	`None`	Merged into the agent's input as `_rerun_context` so it can see what went wrong. Common keys: `previous_output`, `error`, `expected_shape`.
`reason`	string	`"user_request"`	Audit string surfaced in retry telemetry. Common values: `user_request`, `schema_mismatch`, `manager_intervention`, `frontend_rerun_button`, `agent_error`.
`new_input`	dict	`None`	Explicit input. When omitted, the SDK re-uses the agent's most recent stored input automatically.

The feedback is rendered per-agent-type — LLM agents get a correction note in their prompt, human agents see it surfaced in their inbox/channel UI, external connectors receive data._rerun_context, programmatic agents receive input_data._rerun_context. The same dispatch handles every type, so feedback semantics are uniform.

Returns dispatch metadata. The call is fire-and-forget — rerun_agent returns as soon as the dispatch is queued, not when the agent finishes:

{
    "status": "dispatched",
    "agent": "advisor_agent",
    "session_id": "session-1777755-wrh93qrf",
    "input_source": "stored_original" | "new_input",
    "reason": "user_request",
}

How to get the agent's actual output

The right pattern depends on where you called rerun_agent from:

You called it from	How the output reaches you
Inside your own agent / chain step / pre_compose / intelligence function	The re-dispatched agent's output flows back through the normal mesh routing — `can_call`, fan-in, chains all work as if the agent ran for the first time. You don't subscribe to anything. If the rerun is the predecessor of your current agent, your fan-in waits on it naturally.
HTTP REST caller / Studio "Rerun" button	The dispatch is asynchronous. Consume the result via either: (a) Studio's session inspector for the full historical timeline, or (b) the SSE live tail at Studio's session inspector which streams events as they fire.
One-off debug script	Read the agent's stored response via the session API after a short wait, or use the SSE stream above.

The fire-and-forget design is intentional: blocking on the result would force rerun_agent to either (a) tie up the caller's coroutine for minutes (LLM agents can take a long time), or (b) re-implement the mesh's own can_call/fan-in machinery for one specific call. Since both already exist in the mesh, rerun_agent just hands the work to the Manager and gets out of the way.

The same operation is exposed in Studio's Sessions tab as a per-agent rerun button (with optional feedback / new input). Both the Python method and the Studio button route through the same Manager retry path, so feedback rendering is consistent across surfaces.

`await sdk.call_llm(input_data, session_id, model=None, system_prompt=None, temperature=0.1, max_tokens=800, business_context=None, context_parts=None, response_format=None, yields=None, tools=None) -> LLMResponse`

Ad-hoc LLM call with the same provider routing, caching, observability, and event publishing as a full agent. model is required — a silent default would spend a customer's money on a model nobody chose. Use from anywhere — @pre_compose, @chain, @conditional_chain, intelligence functions, or standalone code that needs an LLM round-trip without registering an agent for it (e.g. one-off classification inside a routing decision).

# Quick intent classification inside a chain step
response = await leafmesh.call_llm(
    input_data=user_message,
    session_id=ctx["session_id"],
    model="claude-haiku-4-5",
    system_prompt="Classify intent as one of: refund, complaint, question, other.",
    response_format={"type": "json_schema", "json_schema": {
        "name": "intent", "strict": True,
        "schema": {"type": "object", "properties": {"intent": {"type": "string"}}, "required": ["intent"]}
    }},
)
intent = json.loads(response.content)["intent"]

Argument	Type	Default	Description
`input_data`	str / dict	—	What to send to the LLM
`session_id`	string	—	Required — for conversation tracking
`model`	string	none — required	Any registered model id. Omitting it raises `ModelNotConfiguredError`: "the SDK never chooses a model on your behalf". Pass the calling agent's configured model.
`system_prompt`	string	`None`	System instruction
`temperature`	float	`0.1`	0.0–1.0
`max_tokens`	int	`800`	Response token cap
`response_format`	dict	`None`	JSON Schema for structured output
`yields`	dict	`None`	Output schema (field → type)
`tools`	list[str]	`None`	Tool names to expose
`business_context`	string	`None`	Optional context block
`context_parts`	dict	`None`	Keys: `care`, `sentiment_analysis`, `guardrails`, `flows`

Returns an LLMResponse object (same shape as what intelligence functions receive). Conversation history is recorded automatically, so platform monitoring sees the call like any other turn.

`await leafmesh.trigger_agents(data, session_id=None, target_agents=None) -> None`

From inside an agent's intelligence function, fan an output to one or more downstream agents via event publishing. Uses the contextvar-stored current agent name as the from_agent. Fire-and-forget.

# Inside an intelligence function
async def researcher_agent(llm_response, input_data, context):
    enriched = {"findings": llm_response.content, "ts": time.time()}
    await leafmesh.trigger_agents(
        data=enriched,
        target_agents=["advisor_agent", "auditor_agent"],
    )
    return llm_response

Most agents don't need this — declarative can_call: in YAML handles fan-out automatically. Reach for trigger_agents only when the fan-out targets are computed dynamically.

`await leafmesh.publish_agent_output(from_agent, data, session_id, target_agents=None) -> None`

Lower-level version of trigger_agents — you pass from_agent explicitly instead of relying on contextvars. Useful from background tasks or webhook handlers where the "current agent" isn't in scope.

await leafmesh.publish_agent_output(
    from_agent="external_webhook",
    data={"event": "payment_completed", "amount": 99.0},
    session_id="session_123",
    target_agents=["fulfillment_agent"],
)

Operator Feed

The agent feed is a per-session activity log that auto-captures every agent execution. Operators (or other agents) can attach comments to feed posts — especially @memory-tagged comments, which get injected into the agent's context on future turns.

`await leafmesh.add_feed_comment(post_id, session_id, author, body, *, agent_id=None, is_memory=False, expires_in=None) -> dict`

Add a comment to a feed post. Prefix body with @memory (or pass is_memory=True) to tag it as memory — that memory then surfaces in the agent's prompt context the next time it runs.

# Plain note — visible in the feed UI, not injected into agent context
await leafmesh.add_feed_comment(
    post_id=post["id"],
    session_id=session_id,
    author="ops@leafcraft.ai",
    body="Customer mentioned they're a returning user — see CRM ticket #4421",
)

# Memory-tagged — auto-injected into the agent's next turn
await leafmesh.add_feed_comment(
    post_id=post["id"],
    session_id=session_id,
    author="ops@leafcraft.ai",
    body="@memory Customer is enterprise tier — eligible for white-glove escalation",
    agent_id="advisor_agent",         # Scope to one agent (None = whole session)
    expires_in=3600,                   # TTL in seconds (optional)
)

Argument	Type	Default	Description
`post_id`	string	—	The feed post being commented on
`session_id`	string	—	Session the post belongs to
`author`	string	—	User/operator identifier (free-form)
`body`	string	—	Comment text. `@memory` prefix or `is_memory=True` to tag as memory
`agent_id`	string	`None`	Scope memory to a specific agent (`None` = session-scoped)
`is_memory`	bool	`False`	Explicit memory flag (alternative to `@memory` prefix)
`expires_in`	int	`None`	TTL in seconds for memory-tagged comments

Returns the stored comment dict. Emits feed.comment.added (and feed.comment.memory_tagged if applicable) to streams so Studio's feed updates live.

`await leafmesh.get_feed_comments(post_id) -> list[dict]`

Return every comment on a feed post, oldest-first.

`await leafmesh.get_memory_comments(session_id, *, agent_id=None) -> list[dict]`

Return all memory-tagged comments for a session — audit what notes are currently being injected into agent context. Pass agent_id to filter to agent-scoped memories; omit for session-wide.

`await leafmesh.remove_feed_comment(comment_id, post_id, session_id) -> bool`

Delete a comment. Cleans up the associated memory entry if the comment was memory-tagged. Returns True if anything was deleted. Emits feed.comment.removed to streams.

Session Administration

`await leafmesh.erase_session(session_id, *, actor_id=None, request_id=None) -> bool`

GDPR Art. 17 / ISO 27018 right-to-erasure entry point. Hard-deletes every stored record tied to session_id: session state, conversation history, agent yields, mesh communications, and every channel and connector record for the session.

deleted = await leafmesh.erase_session(
    session_id="session_001",
    actor_id="user_42",
    request_id="gdpr-2026-03-14-001",   # Audit trail
)

Emits a session.erased audit event with the supplied actor_id / request_id so the deletion is auditable (SOC 2 CC7.2). Returns True if anything was erased.

`leafmesh.cancel_dispatch(session_id, agent_name) -> bool`

Synchronous (no await). Super-agent v3 cancellation — sets the cancel flag for an in-flight super-agent dispatch. The orchestrator checks the flag between phases and unwinds gracefully: current phase finishes, plan state is preserved and marked cancelled, the response returns _cancelled: true.

cancelled = leafmesh.cancel_dispatch("session_001", "researcher_agent")
# cancelled = True if there was an in-flight dispatch to cancel

A subsequent sdk.resume_super_agent call picks the plan back up at the next pending step. Safe to call from any context — request handlers, scheduled jobs, Manager interventions.

Agent Registration

Intelligence Functions

Register a post-LLM intelligence function. The function name must match the agent name in YAML — leafmesh.discover() finds them automatically:

async def my_agent(llm_response, input_data, context):
    # 3-param: receives LLM response, input, and context
    return llm_response

`@pre_compose(...)`

Standalone decorator for pre-LLM processing. Runs deterministic Python before the LLM call:

from leafmesh import pre_compose

@pre_compose(input_processor=lambda data: {**data, "extra": "data"})
async def my_agent(llm_response, input_data, context):
    return llm_response

`leafmesh.discover(directory=".", pattern="*_agent.py", recursive=False)`

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

leafmesh.discover("./agents", pattern="*_agent.py", recursive=True)

`sdk.agent_registry.register_human_agent(name, human_interface_handler=None, **config)`

Register a human-powered agent. The handler is called when the mesh wants the human to respond; if omitted, the human-agent system falls back to channel adapters (Slack/Telegram/Teams/Email/webhook) wired in the agent's YAML config.

def slack_relay(input_data, context):
    slack_client.post_message(channel="#ops", text=f"Need a decision: {input_data}")
    # Response arrives via webhook → mesh routes it back into the session
    return None  # async — response delivered separately

sdk.agent_registry.register_human_agent("ops_reviewer", slack_relay,
                                        timeout_seconds=600)

It lives on the agent registry, not on the SDK object — sdk.register_human_agent(...) does not exist. To swap the handler on a seat that is already declared, use sdk.agent_registry.update_human_interface_handler("client", handler).

YAML alternative: set agent_type: human on the agent in YAML and the SDK auto-registers it on start(). Programmatic registration is for cases where the handler is closure-bound to runtime state that YAML can't capture.

`leafmesh.register_llm_provider(name, provider_class, model_prefixes=None, **config)`

Register a custom LLM provider (subclass of leafmesh.LLMProvider) for use with any agent. Useful for proprietary models, internal model gateways, or models the SDK doesn't ship with.

from leafmesh import LLMProvider, LLMRequest, LLMResponse

class MyCustomProvider(LLMProvider):
    _provider_name = "my-custom"
    async def generate(self, request: LLMRequest) -> LLMResponse:
        # ... call your model ...
        return LLMResponse(content=text, model=request.model, usage={...},
                           provider="my-custom", success=True)
    def is_available(self) -> bool:
        return self._api_key is not None

leafmesh.register_llm_provider(
    name="my-custom",
    provider_class=MyCustomProvider,
    model_prefixes=["mycustom/"],
    api_key=os.environ["MY_CUSTOM_KEY"],
)

# Now any agent with `model: mycustom/foo-7b` routes to MyCustomProvider.

If the executor is already started, the provider is added live. Otherwise it joins at next start().

`await leafmesh.register_background_task(task)`

Register an asyncio.Task so the SDK tracks it through shutdown. On leafmesh.stop(), the SDK waits up to ~120s for registered tasks to complete before forcing cancellation. Use for any agent-spawned background work that should drain cleanly.

task = asyncio.create_task(my_long_running_thing())
await leafmesh.register_background_task(task)

Function Composition

`leafmesh.chain(*functions)`

Chain multiple functions into a pipeline:

pipeline = leafmesh.chain(validate, process, format_output)

`leafmesh.conditional_chain(condition_func, *functions)`

Chain that only executes if condition is met.

`leafmesh.chain_with_results(*functions)`

Chain that collects results from all functions.

`leafmesh.compose(**processors)`

Composition decorator for modular components.

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.

Analytics & Monitoring

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

Analytics are also available via the REST API (see below).

REST API Endpoints (Port 18820)

LeafMesh API server runs at http://127.0.0.1:18820. Override the port with the LEAFMESH_API_PORT environment variable. The separate evolution/health API defaults to 19820 and is overridable with LEAFMESH_EVOLUTION_API_PORT (precedence: env var → YAML → default).

Method	Endpoint	Description
GET	`/health`	System health check
POST	`/api/mesh/request`	Submit a request via an entry point
GET	`/api/mesh/entry_points`	List configured entry points
POST	`/api/mesh/stream`	Stream a mesh request (SSE)
GET	`/api/mesh/stream/{session_id}`	Re-attach to a session's live reply stream (SSE) — resumes from any worker/replica
GET	`/api/followups`	List pending self-scheduled follow-ups (`?session_id=`)
DELETE	`/api/followups/{id}`	Cancel a pending follow-up (`?session_id=`)
GET	`/api/budgets`	Per-agent monthly budget + current-month utilisation
GET	`/api/budgets/{agent}`	One agent's budget + utilisation
POST	`/webhook/{entry_point}`	Webhook ingress — new task or human response
POST	`/callback/{agent_name}`	External-system callback for connectors with `after_handoff: wait` or `move_on` (2.4.508; was `mode: callback`)
POST	`/callback/calls/{framework}`	A phone-call platform's "call finished" webhook (2.4.508)

Broker connections defined under top-level brokers: are referenced by agent.listen_events and shared across any number of subscriptions. Secrets such as sasl_password, aws_secret_access_key, aws_session_token, and password are redacted at rest in any operational output.

Next Steps

Agent Classes — Agent API reference
Configuration API — Config reference
Event System — Events and real-time monitoring

LeafMesh — LeafMesh class reference