Agent Communication

How agents exchange data through the managed mesh in LeafMesh.

Yields-Based Data Flow

Agents communicate through structured yields. When Agent A calls Agent B, Agent A's yields become Agent B's input_data:

agents:
  classifier:
    name: "classifier"
    yields:
      category: "string"
      confidence: "number"
    can_call:
      - agent: "handler"
        condition: "confidence > 0.8"

  handler:
    name: "handler"
    yields:
      result: "string"

async def handler(llm_response, input_data, context):
    # input_data = classifier's yields: {category: "...", confidence: 0.9}
    category = input_data.get("category")
    return llm_response

The contract travels, never the ancestry

Declared yields are the handoff contract, and the mesh enforces it end to end:

One-level handoffs. A raw handoff carries the sender's own output only. Context the sender itself received — its upstream results, fan-in accumulations, chain history — is stripped before dispatch, so payloads never compound hop over hop.
@compose / @chain shapers are exempt. A shaped payload is your contract; the mesh never edits it.
upstream_yields is exactly that. Fan-in targets receive each contributor's declared yields — parsed, contract-shaped — under upstream_yields, alongside the raw contributions under upstream_results. This holds for every agent type, including super-agents (whose synthesis honors the same contract).
Rendered once. In the downstream agent's prompt each contributor appears exactly once — the yields view wins; raw output renders only for a contributor that has no yields entry, so nothing is ever silently dropped.

`wait_for` is a gate

When an agent declares wait_for, that expression is a hard gate on its dispatch — not a hint:

advisor_agent:
  wait_for: processor_agent AND researcher_agent
  wait_for_timeout: 300     # safety net, seconds

While the join is pending, every dispatch to the agent is held as a contribution — including from agents that are not named in the expression. Their data is banked and rides along in upstream_results, but only the expression's own terms open the gate.
Once the join has completed, later calls to the same agent are ordinary chain dispatches again.
wait_for_timeout is the safety net: if the expression can't be satisfied in time (a contributor died, a network outage), the agent fires with whatever arrived, explicitly reporting the skipped and failed contributors.

Mind the boolean shape: processor AND researcher OR (fallback) is satisfied the moment the fallback lands. If the intent is "wait for the real researcher", say processor AND researcher and let the timeout cover failure.

Communication Patterns

Sequential Chain

# A → B → C (each calls the next)
agents:
  step_1:
    name: "step_1"
    can_call:
      - agent: "step_2"
        condition: "status == 'ready'"
  step_2:
    name: "step_2"
    can_call:
      - agent: "step_3"
        condition: "status == 'ready'"
  step_3:
    name: "step_3"
    yields:
      final_result: "string"

Fan-Out (One to Many)

agents:
  coordinator:
    name: "coordinator"
    can_call:
      - agent: "analyst_a"
        condition: "true"
        call_immediately: true
      - agent: "analyst_b"
        condition: "true"
        call_immediately: true

Or via direct mesh calls in an intelligence function:

async def coordinator(llm_response, input_data, context):
    import asyncio
    session_id = context.get("session_id", "default")

    results = await asyncio.gather(
        leafmesh.mesh_call("coordinator", "analyst_a", input_data, session_id=session_id),
        leafmesh.mesh_call("coordinator", "analyst_b", input_data, session_id=session_id)
    )

    return {"analyst_a": results[0], "analyst_b": results[1]}

Fan-In (`wait_for`)

wait_for is the join side of fan-out: an agent declares a boolean expression over its upstream contributors and only runs once that expression is satisfied.

agents:
  aggregator:
    name: "aggregator"
    wait_for: "analyst_a AND (analyst_b OR analyst_c)"
    wait_for_timeout: 120          # seconds (grace window, optional)

Required vs optional contributors. A bare name is required — the target waits for it. Suffix a name with ? to make it optional:

wait_for: "analyst_a AND analyst_b?"

Here analyst_a must contribute; analyst_b is best-effort.

Semantics worth knowing:

Optional siblings are still waited for while in flight. If an optional contributor was dispatched off the same upstream output as a required one, the target won't fire the instant the required arrives — it waits for the optional too, so the target doesn't run without data it was configured to consider. The optional only stops blocking once it finishes, fails, or the timeout grace expires.
Late arrivals are dropped. Once a fan-in has been satisfied and the target has fired, any contributor that arrives afterward is discarded — the target is not re-run.
Timeout grace runs the target with what arrived. If only optional contributors are still missing when the grace window (wait_for_timeout) expires, the target runs with whatever has arrived. If a required contributor's path is dead (it failed or was skipped with no alternative), the expression becomes unsatisfiable and the join escalates to the Manager instead of hanging forever.

Conditional Branching

agents:
  router:
    name: "router"
    yields:
      request_type: "string"
    can_call:
      - agent: "technical_support"
        condition: "request_type == 'technical'"
      - agent: "billing_support"
        condition: "request_type == 'billing'"
      - agent: "general_support"
        condition: "request_type not in ['technical', 'billing']"

Human Gating

agents:
  processor:
    name: "processor"
    can_call:
      - agent: "human_reviewer"
        condition: "risk_level == 'high'"

  human_reviewer:
    name: "human_reviewer"
    agent_type: "human"
    can_call:
      - agent: "executor"
        condition: "approval == 'approved'"

Direct Mesh Calls

Intelligence functions can make direct mesh calls for dynamic routing:

async def dynamic_router(llm_response, input_data, context):
    session_id = context.get("session_id", "default")
    target = llm_response.get("recommended_agent", "default_handler")

    result = await leafmesh.mesh_call(
        "dynamic_router", target, input_data, session_id=session_id
    )
    return result

Session Context in Communication

All agents in the same session share state automatically through the mesh. When Agent A calls Agent B, Agent A's yields become Agent B's input_data. This is the primary mechanism for passing data between agents.

Yields from each agent are auto-persisted and auto-passed to downstream agents. You do not need to manage this manually. Browse session state in Studio's Sessions tab.

Next Steps

Coordination Patterns — Automatic oversight and intervention
Mesh Architecture — Routing infrastructure
Message Routing — Condition evaluation details

LeafMesh — Agent-to-agent data exchange