Decorators

Decorators are how you write agent logic in Python. They wrap your agent functions with pre-processing, post-processing, branching, and handoff-shaping behavior -- all without touching the YAML configuration.

Every agent function uses the same three-parameter signature:

async def my_agent(llm_response, input_data, context):
    # llm_response: parsed LLM output (None for programmatic agents)
    # input_data:   data from the caller or upstream agent
    # context:      {"session_id": "...", "agent_name": "...", ...}
    return {"key": "value"}  # must match the agent's yields schema

Decorators modify what happens before or after that function runs. Import them directly from the leafmesh package:

from leafmesh import pre_compose, chain, compose, conditional_chain, chain_with_results

The `context` pool — shared across every stage

The context dict is the same object across the turn's in-process stages: @pre_compose receives it, your agent function receives it, and every @chain step receives it — one shared scratchpad that accumulates as the turn progresses. Whatever an earlier stage writes into it is still there for a later stage.

@compose is the exception. Handoff shapers run later, at publish time in the mesh, and receive a fresh minimal ctx (session_id, tenant_id, from_agent) — not the turn's context. If a shaper needs pool-derived data, put it into the result in your function or a @chain step; the shaper reads it from there.

def load_account(input_data, context):
    context["account"] = fetch_account(input_data["account_id"])  # write to the pool
    return {"user_message": input_data["message"]}                # this is what the LLM sees

@pre_compose(context_processor=load_account)
async def support_agent(llm_response, input_data, context):
    account = context["account"]        # read it back after the LLM answered
    return {"reply": llm_response, "tier": account["tier"], "status": "answered"}

It's a side channel — not prompt content

The LLM prompt is built from input_data and the agent's messages, not from arbitrary context keys. So data you park in context under your own key is not shown to the model. That's what makes this pattern possible:

Pull a full dataset in @pre_compose, send only a subset to the LLM, then rejoin the full set in @chain.

# agency/enrichment_agent.py
from leafmesh import pre_compose, chain

def stage_records(input_data, context):
    records = fetch_records(input_data["query"])   # the full set — many fields each
    context["records"] = records                   # stash the WHOLE thing in the pool
    # send the model only the slim view it needs to reason over:
    return {"items": [{"id": r["id"], "label": r["label"]} for r in records]}

def rejoin(result, context):
    # the LLM ranked the slim items; now stitch the full records back on
    ranked_ids = result.get("ranked_ids", [])
    by_id = {r["id"]: r for r in context["records"]}
    result["ranked_records"] = [by_id[i] for i in ranked_ids if i in by_id]
    return result

@chain(rejoin)
@pre_compose(input_processor=stage_records)
async def enrichment_agent(llm_response, input_data, context):
    # the model saw only id + label, and returned an ordering
    return {"ranked_ids": llm_response.get("ranked_ids", []), "status": "ranked"}

The model reasons over the slim view; the full records never enter the prompt but are rejoined afterward from the pool. The rejoin is your code — the pool holds the data and threads it through; you write the merge in a @chain step.

Reserved keys that do reach the prompt

A fixed set of context keys is read into the prompt on purpose — don't use them to hide data. The content-bearing ones are:

prepared_data, agent_memory, agent_knowledge, conversation_summary, conversation_history, previous_input, previous_output, agent_data.

Note prepared_data in particular: @pre_compose processors write there by default, and it is surfaced to the model. To keep something out of the prompt, stash it under a key of your own choosing (like context["records"] above), not one of these.

`context` is per-turn; `upstream_yields` is cross-agent

The pool lives for one agent's turn and is rebuilt on the next call — it is not how data travels between agents. When an agent hands off, what the next agent receives is its yields, delivered as input_data["upstream_yields"][caller_name]. Use context to carry data within a turn (pre_compose → chain → compose); use yields to carry data to the next agent.

@pre_compose

Runs processor functions before the LLM call to prepare data. Each processor enriches the context that the LLM sees before generating its response.

When to use it

Use @pre_compose on LLM agents when you need to clean user input, load business context, or attach auxiliary data before the prompt is sent to the model.

Function signature

@pre_compose(
    context_processor=callable,   # Prepares business/domain context
    input_processor=callable,     # Cleans and normalizes user input
    others_processor=callable,    # Loads auxiliary data (flags, config, etc.)
)
async def my_agent(llm_response, input_data, context):
    ...

Each processor receives (input_data, context) and returns a dict. The returned values are merged into the prompt context that the LLM sees — your agent function receives the enriched context and input_data and can read whatever it returned.

Processor	Role
`context_processor`	Adds business / domain context
`input_processor`	Cleans and normalizes user input
`others_processor`	Attaches supplementary data (flags, config, etc.)

Code example

# agency/greeter_agent.py
from leafmesh import pre_compose


def process_context(input_data, context):
    """Enrich the prompt with domain info."""
    return {
        "domain": "task_management",
        "session_id": context.get("session_id", "unknown"),
    }


def clean_input(input_data, context):
    """Normalize the user's raw message."""
    raw = input_data.get("message", "")
    return {"user_message": raw.strip()}


def load_extras(input_data, context):
    """Attach feature flags or supplementary data."""
    return {"timestamp_requested": "message" in input_data}


@pre_compose(
    context_processor=process_context,
    input_processor=clean_input,
    others_processor=load_extras,
)
async def greeter_agent(llm_response, input_data, context):
    # `context` and `input_data` are pre-enriched by the processors above;
    # the LLM has already seen them when it produced `llm_response`.
    return {
        "greeting": llm_response if llm_response else "Hello!",
        "user_input": input_data.get("message", ""),
        "detected_sentiment": "neutral",
        "domain": context.get("domain", "general"),
        "status": "greeted",
    }

Corresponding YAML

greeter_agent:
  name: "greeter_agent"
  agent_type: "llm"
  model: "gpt-4o-mini"
  temperature: 0.7
  max_tokens: 1000

  prompt: |
    You are a friendly assistant. Given the user's message,
    greet them warmly and summarize what they need help with.
    Return a structured JSON with keys:
    "greeting", "user_need", "suggested_actions"

  context_parts:
    care: |
      Always respond with empathy and warmth.
    guardrails: |
      Never share internal system details.

  yields:
    greeting: "string"
    user_input: "string"
    detected_sentiment: "string"
    status: "string"

  inputs:
    message: "string"

  can_call:
    - agent: "processor_agent"

Execution order

context_processor, input_processor, others_processor run (before the LLM call)
context_parts (from YAML) are injected as system messages
LLM generates a response
Your agent function runs with the LLM output and prepared data

@chain

Runs a sequential pipeline of functions after your agent function returns. Each step receives the previous step's output and transforms it further.

When to use it

Use @chain when you need to validate, enrich, or format your agent's output in discrete steps before it is passed downstream.

Function signature

@chain(step_one, step_two, step_three)
async def my_agent(llm_response, input_data, context):
    ...

Each chain step receives (result, context) and returns the transformed result. Steps run in order.

Code example

# agency/fallback_researcher_agent.py
from leafmesh import chain


def enrich_findings(result, context):
    """Add analysis metadata to findings."""
    findings = result.get("findings", [])
    result["analysis"] = {
        "total_findings": len(findings),
        "confidence_score": 0.6,
        "data_quality": "template",
    }
    result["report"] = {
        "title": "Fallback Research Findings",
        "confidence": 0.6,
        "finding_count": len(findings),
        "formatted": True,
    }
    return result


@chain(enrich_findings)
async def fallback_researcher_agent(llm_response, input_data, context):
    upstream_yields = input_data.get("upstream_yields", {})
    processor_yields = upstream_yields.get("processor_agent", {})

    calling_data = input_data.get("calling_agent_response", {})
    processed_items = calling_data.get("processed_items", [])

    findings = []
    for item in processed_items:
        findings.append({
            "source": "template",
            "content": item.get("item", ""),
            "recommendation": f"Review and address: {item.get('item', 'unknown')}",
        })

    return {
        "findings": findings,
        "query": input_data.get("research_topic", ""),
        "upstream_item_count": processor_yields.get("item_count", 0),
        "status": "researched",
    }

Corresponding YAML

fallback_researcher_agent:
  name: "fallback_researcher_agent"
  agent_type: "programmatic"
  description: "Fast template-based research fallback"
  communication_type: "chain"

  yields:
    findings: "list"
    query: "string"
    analysis: "object"
    report: "object"
    status: "string"

  inputs:
    processed_data: "object"

  can_call:
    - agent: "advisor_agent"

Execution order

Your agent function runs
step_one(result, context) runs
step_two(result, context) runs (receives step_one's output)
Final result is passed to can_call targets

@compose

Shapes the output payload differently for each downstream agent. When your agent hands off to multiple targets via can_call, each target can receive a tailored subset of the data instead of the full result.

When to use it

Use @compose when one agent fans out to multiple downstream agents that each need different fields. Targets not listed in @compose receive the full unmodified result.

Function signature

@compose(
    target_agent_name=shaper_function,
    another_agent_name=shaper_function,
)
async def my_agent(llm_response, input_data, context):
    ...

Each shaper function receives (result, context) and returns the shaped payload for that specific target.

Code example

from leafmesh import compose


def shape_report(data, context):
    """Shape output for a reporting agent."""
    return {
        "title": "Advisory Report",
        "recommendations": data.get("recommendations", []),
        "risk_assessment": data.get("risk_assessment", ""),
        "priority_score": data.get("priority_score", 0),
    }


def shape_alerts(data, context):
    """Shape output for an alerting agent."""
    score = data.get("priority_score", 0)
    return {
        "alert_level": "high" if score > 0.7 else "medium" if score > 0.4 else "low",
        "summary": data.get("risk_assessment", ""),
        "action_required": score > 0.5,
    }


@compose(report_agent=shape_report, alert_agent=shape_alerts)
async def triage_agent(llm_response, input_data, context):
    return {
        "recommendations": ["Fix the login flow", "Update onboarding"],
        "risk_assessment": "elevated -- login errors spiking",
        "priority_score": 0.8,
        "raw_data": {"error_count": 47, "affected_users": 230},
        "status": "triaged",
    }

With this setup:

report_agent receives {"title": "Advisory Report", "recommendations": [...], ...} (clean report)
alert_agent receives {"alert_level": "high", "summary": "...", "action_required": true} (alert payload)
Any other can_call target receives the full result dict

Corresponding YAML

triage_agent:
  name: "triage_agent"
  agent_type: "llm"
  model: "gpt-4o-mini"
  prompt: |
    Analyze the incoming data and produce recommendations.

  yields:
    recommendations: "list"
    risk_assessment: "string"
    priority_score: "number"
    status: "string"

  can_call:
    - agent: "report_agent"
    - agent: "alert_agent"
    - agent: "archive_agent"   # receives full result (no shaper defined)

@conditional_chain

Runs a chain of functions only when a condition is met. If the condition returns False, the agent's result passes through unchanged.

When to use it

Use @conditional_chain when your post-processing logic should only run in certain cases -- for example, escalating urgent items or applying special formatting when thresholds are exceeded.

Function signature

@conditional_chain(condition_function, step_one, step_two)
async def my_agent(llm_response, input_data, context):
    ...

The condition function receives (result, context) and returns True or False. If True, the chain steps run sequentially. If False, the result passes through unmodified.

Code example

# agency/processor_agent.py
from leafmesh import conditional_chain


def handle_urgent(result, context):
    """Flag urgent items for immediate attention."""
    urgent = [i for i in result.get("processed_items", []) if i["category"] == "urgent"]
    result["urgent_items"] = urgent
    result["alert"] = f"{len(urgent)} urgent item(s) flagged"
    return result


def handle_normal(result, context):
    """Default handling when nothing is urgent."""
    result["alert"] = "No urgent items"
    return result


@conditional_chain(
    lambda result, ctx: any(
        i["category"] == "urgent" for i in result.get("processed_items", [])
    ),
    handle_urgent,
    handle_normal,
)
async def processor_agent(llm_response, input_data, context):
    upstream_yields = input_data.get("upstream_yields", {})
    greeter_yields = upstream_yields.get("greeter_agent", {})
    user_input = input_data.get("user_input", "")

    raw_items = [item.strip() for item in user_input.split(",") if item.strip()]

    categorized = []
    for item in raw_items:
        categorized.append({
            "item": item,
            "category": "urgent" if any(w in item.lower() for w in ["asap", "urgent", "now"]) else "normal",
        })

    return {
        "processed_items": categorized,
        "item_count": len(categorized),
        "status": "processed",
    }

Corresponding YAML

processor_agent:
  name: "processor_agent"
  agent_type: "programmatic"
  description: "Processes greeter output into actionable items"
  communication_type: "chain"
  scaling:
    runs: {max: 3}

  yields:
    processed_items: "list"
    item_count: "number"
    status: "string"
    alert: "string"

  inputs:
    greeter_result: "object"

  can_call:
    - agent: "researcher_agent"
      condition: "calling_agent_response.item_count > 0"
    - agent: "advisor_agent"
      condition: "calling_agent_response.item_count > 0"

Execution order

Your agent function runs
The condition function evaluates against the result
If True: chain steps run sequentially
If False: result passes through unmodified
Final result is passed to can_call targets

@chain_with_results

Like @chain, but accumulates every intermediate result into a list. The final return value is a dict containing the main result, all chain step results, and the final transformed result.

When to use it

Use @chain_with_results when you need visibility into what each chain step produced -- useful for auditing, debugging, or when downstream agents need to inspect intermediate transformations.

Function signature

@chain_with_results(step_one, step_two, step_three)
async def my_agent(llm_response, input_data, context):
    ...

Each chain step receives (result, context) and returns its transformed result.

Return structure

{
    "main_result": ...,      # Your agent function's return value
    "chain_results": [...],  # List of each step's return value
    "final_result": ...,     # The last step's return value
}

Code example

# agency/researcher_agent.py
from leafmesh import chain_with_results


def analyze_data(result, context):
    """Step 1: Compute analysis metrics."""
    items = result.get("findings", [])
    result["analysis"] = {
        "total_findings": len(items),
        "confidence_score": 0.85 if items else 0.1,
        "data_quality": "high" if len(items) >= 3 else "limited",
    }
    return result


def format_findings(result, context):
    """Step 2: Structure into a formal report."""
    analysis = result.get("analysis", {})
    result["report"] = {
        "title": "Research Findings",
        "confidence": analysis.get("confidence_score", 0),
        "quality": analysis.get("data_quality", "unknown"),
        "finding_count": analysis.get("total_findings", 0),
        "formatted": True,
    }
    return result


@chain_with_results(analyze_data, format_findings)
async def researcher_agent(llm_response, input_data, context):
    memory_posts = context.get("memory_posts", [])
    prior_findings = sum(1 for p in memory_posts if p.get("role") == "assistant")

    findings = []
    if llm_response:
        findings.append({"source": "llm", "content": llm_response})

    calling_data = input_data.get("calling_agent_response", {})
    if calling_data.get("processed_items"):
        for item in calling_data["processed_items"]:
            findings.append({"source": "upstream", "content": item.get("item", "")})

    return {
        "findings": findings,
        "query": input_data.get("research_topic", ""),
        "prior_research_sessions": prior_findings,
        "status": "researched",
    }

Corresponding YAML

researcher_agent:
  name: "researcher_agent"
  agent_type: "llm"
  model: "gpt-4o-mini"
  temperature: 0.5
  max_tokens: 1500
  reasoning: true
  memory: true
  memory_limit: 10

  prompt: |
    You are a research analyst. Analyze the provided data thoroughly.
    Use available tools to gather additional information.
    Return structured findings with confidence scores.

  yields:
    findings: "list"
    query: "string"
    analysis: "object"
    report: "object"
    status: "string"

  inputs:
    research_topic: "string"
    processed_data: "object"

  tools: ["math_eval", "timestamp"]
  tool_categories: ["data"]
  tool_choice: "auto"
  allow_parallel_tool_calls: true

  can_call:
    - agent: "advisor_agent"

Combining Decorators

Decorators can be stacked. The most common combination is @chain + @compose -- chain post-processes the result, then compose shapes it per target.

@chain + @compose

# agency/advisor_agent.py
from leafmesh import chain, compose


def validate_inputs(result, context):
    """Chain step 1: Ensure all required fields are present."""
    if not result.get("recommendations"):
        result["recommendations"] = ["No specific recommendations available"]
    if not result.get("risk_assessment"):
        result["risk_assessment"] = "Unable to assess -- insufficient data"
    result["validated"] = True
    return result


def score_priorities(result, context):
    """Chain step 2: Score and rank recommendations."""
    recs = result.get("recommendations", [])
    result["priority_score"] = min(1.0, len(recs) * 0.2) if recs else 0.0
    result["next_steps"] = recs[:3] if recs else ["Gather more data"]
    result["scored"] = True
    return result


def shape_report(data, context):
    """Compose: shape for reporting systems."""
    return {
        "title": "Advisory Report",
        "recommendations": data.get("recommendations", []),
        "risk_assessment": data.get("risk_assessment", ""),
        "priority_score": data.get("priority_score", 0),
    }


def shape_alerts(data, context):
    """Compose: shape for alerting systems."""
    score = data.get("priority_score", 0)
    return {
        "alert_level": "high" if score > 0.7 else "medium" if score > 0.4 else "low",
        "summary": data.get("risk_assessment", ""),
        "action_required": score > 0.5,
    }


@chain(validate_inputs, score_priorities)
@compose(report_agent=shape_report, alert_agent=shape_alerts)
async def advisor_agent(llm_response, input_data, context):
    upstream_yields = input_data.get("upstream_yields", {})
    processor_yields = upstream_yields.get("processor_agent", {})
    researcher_yields = upstream_yields.get("researcher_agent", {})

    processed_items = processor_yields.get("processed_items", [])
    findings = researcher_yields.get("findings", [])

    recommendations = []
    if processed_items:
        urgent = [i for i in processed_items if i.get("category") == "urgent"]
        if urgent:
            recommendations.append(f"Address {len(urgent)} urgent item(s) immediately")
        recommendations.append(f"Review {len(processed_items)} processed items")
    if findings:
        recommendations.append(f"Incorporate {len(findings)} research finding(s)")

    risk = "low"
    if any(i.get("category") == "urgent" for i in processed_items):
        risk = "elevated -- urgent items detected"

    return {
        "recommendations": recommendations,
        "risk_assessment": risk,
        "status": "advised",
    }

When stacked, the execution order is:

Your agent function runs
@compose shapers are registered (they run at handoff time, not here)
@chain steps run: validate_inputs then score_priorities
On handoff, @compose shapers transform the final result per target

Decorator Quick Reference

Decorator	Runs	Purpose	Chain steps receive
`@pre_compose`	Before LLM call	Prepare input data and context	`(input_data, context)`
`@chain`	After agent function	Sequential post-processing	`(result, context)`
`@compose`	At handoff time	Per-target payload shaping	`(result, context)`
`@conditional_chain`	After agent function (conditionally)	Branching post-processing	`(result, context)`
`@chain_with_results`	After agent function	Sequential post-processing with intermediate results	`(result, context)`

No Decorator Needed

Not every agent needs a decorator. If your agent function simply processes data and returns a result, skip the decorator entirely:

# agency/scheduler_agent.py
from datetime import datetime, timezone


async def scheduler_agent(llm_response, input_data, context):
    """Runs on a cron schedule. No decorator needed."""
    now = datetime.now(timezone.utc)
    return {
        "report": f"Scheduled check completed at {now.isoformat()}",
        "generated_at": now.isoformat(),
        "checks": {
            "system_status": "healthy",
            "scheduled_run": True,
        },
        "status": "report_generated",
    }

Auto-discovery matches the function name scheduler_agent to the YAML agent config -- no import from leafmesh required.

Next Steps

Agent Development -- Full development workflow and testing patterns
Agent Types -- LLM, programmatic, human, and external agents
Scheduling -- Cron-based agent scheduling

LeafMesh -- Build, test, iterate

Decorators

The context pool — shared across every stage

It's a side channel — not prompt content

Reserved keys that do reach the prompt

context is per-turn; upstream_yields is cross-agent

@pre_compose

When to use it

Function signature

Code example

Corresponding YAML

Execution order

@chain

When to use it

Function signature

Code example

Corresponding YAML

Execution order

@compose

When to use it

Function signature

Code example

Corresponding YAML

@conditional_chain

When to use it

Function signature

Code example

Corresponding YAML

Execution order

@chain_with_results

When to use it

Function signature

Return structure

Code example

Corresponding YAML

Combining Decorators

@chain + @compose

Decorator Quick Reference

No Decorator Needed

Next Steps

The `context` pool — shared across every stage

`context` is per-turn; `upstream_yields` is cross-agent