Guides

Register agents and compose workflows

By the end of this page you will know all three ways to register an agent, when each one earns its keep, and how to chain agents into sequential pipelines and parallel DAGs — with conditional branching that is safe by construction.

One concept underpins everything here: AgentMetadata. However you register an agent — decorator, builder, or YAML — the result is the same record: a safe_name (the unique key, also used as the MCP tool name), a Replicate model path, a description, tags, streaming support, cost and latency estimates, and an optional JSON Schema for input validation. The registry stores these records by safe_name; the executor, the router, and the MCP server all read from the same registry. Pick the registration style that fits your code, not the other way around.

Steps

1. 1

   Register agents

   Decorator for code you own; AgentBuilder for programmatic, config-driven registration.

2. 2

   Compose a sequential workflow

   WorkflowBuilder.then() chains steps; input_map and condition control data flow.

3. 3

   Go parallel with a DAG

   AgentWorkflow runs same-level nodes concurrently — the critic/advocate pattern.

4. 4

   Run and checkpoint from the CLI

   workflows run executes step-by-step, with --checkpoint-dir and --resume-from for recovery.

1. Register agents: decorator or builder

The @agent decorator is the declarative style — registration is a side effect of import, and the function body builds the payload sent to the Replicate model. Every keyword maps directly to an AgentMetadata field:

from replicate_mcp import agent

@agent(
    model="meta/meta-llama-3-70b-instruct",   # defaults to the function name
    description="Drafts marketing copy",       # defaults to the docstring
    tags=["text", "marketing"],
    supports_streaming=True,
    estimated_cost=0.002,
    input_schema={
        "type": "object",
        "properties": {"prompt": {"type": "string"}},
        "required": ["prompt"],
    },
    # registry=...  # optional: target a specific AgentRegistry
)
def drafter(prompt: str) -> dict:
    return {"prompt": prompt}

AgentBuilder produces the identical metadata through method chaining. Reach for it when agent definitions come from configuration, a loop, or anywhere a decorator cannot go. Every setter returns the builder; finish with .build() to get the metadata without registering, or .register() to do both:

from replicate_mcp.sdk import AgentBuilder

meta = (
    AgentBuilder("critic")
    .model("meta/llama-3.1-405b-instruct")
    .description("Scores a draft for clarity and tone")
    .tag("text", "review")
    .streaming(False)
    .estimated_cost(0.004)
    .avg_latency(6000)
    .input_schema({"type": "object", "properties": {"prompt": {"type": "string"}}})
    .register()   # or .build() to construct without registering
)

Rule of thumb: decorator when the agent lives next to application code you import anyway; builder when names and models are only known at runtime. Both call register_or_update() under the hood, so re-importing a module never raises — the stricter register() on the registry itself is what raises DuplicateAgentError on a name collision.

2. Compose a sequential workflow

WorkflowBuilder chains agents into a pipeline where each step's output feeds the next step's input. The method is .then() — each call appends a step by the agent's safe_name, with two optional keywords: input_map remaps keys from the previous step's output (or the initial workflow input) into this step's input, and condition is a guard expression that skips the step when it evaluates false.

from replicate_mcp.sdk import WorkflowBuilder, register_workflow

wf = (
    WorkflowBuilder("content_creation_pipeline")
    .description("Draft copy, review it, illustrate it")
    .then("drafter", input_map={"prompt": "topic"})
    .then("critic", input_map={"prompt": "output"})
    .then("flux_pro", input_map={"prompt": "output"}, condition="len(output) > 40")
    .build()
)
register_workflow(wf)   # now runnable by name from the CLI

Conditions are written in a deliberately restricted expression language (replicate_mcp.dsl.SafeEvaluator), not full Python. Allowed: literals, arithmetic, comparisons (==, <, in, …), boolean logic, subscripts like data["key"], conditional expressions, and a fixed set of safe builtins (len, min, max, sum, …). Forbidden: import, eval, lambdas and any function definition, and access to dunder attributes like __class__. The evaluator parses and validates every expression against an AST allow-list before anything executes, so an unsafe condition fails with UnsafeExpressionError instead of running.

3. Go parallel with a DAG

WorkflowBuilder pipelines are strictly sequential. For fan-out, use AgentWorkflow from replicate_mcp.agents.composition: a validated DAG where nodes at the same topological level run concurrently in an anyio task group. The classic use is the critic/advocate pattern — two reviewers reading the same draft at the same time:

from replicate_mcp.agents.composition import AgentNode, AgentWorkflow, WorkflowEdge

wf = AgentWorkflow(name="review_board", description="Parallel critique")
wf.add_agent("drafter", AgentNode(model_id="meta/meta-llama-3-70b-instruct", role="orchestrator"))
wf.add_agent("critic", AgentNode(model_id="meta/llama-3.1-405b-instruct", role="critic"))
wf.add_agent("advocate", AgentNode(model_id="meta/meta-llama-3-70b-instruct", role="critic"))
wf.add_edge(WorkflowEdge(from_agent="drafter", to_agent="critic"))
wf.add_edge(WorkflowEdge(from_agent="drafter", to_agent="advocate"))

wf.execution_levels()   # [['drafter'], ['advocate', 'critic']] — level 2 fans out

add_edge runs cycle detection on every call, so an edge that would make the graph loop is rejected immediately. Edges also accept transform and condition callables applied between levels.

The same agents/edges shape exists declaratively in YAML — this is examples/workflows/content_pipeline.yaml from the repository, where transform and condition are names registered in the TransformRegistry:

name: content_creation_pipeline
description: Generate, critique, and refine marketing content

agents:
  - id: ideator
    model: anthropic/claude-4.5-sonnet
    role: orchestrator
    streaming: true
  - id: image_gen
    model: black-forest-labs/flux-1.1-pro
    role: specialist
  - id: critic
    model: meta/llama-3.1-405b-instruct
    role: critic
    fallback: anthropic/claude-3.5-sonnet

edges:
  - from: ideator
    to: image_gen
    transform: extract_prompt            # registered in TransformRegistry
  - from: image_gen
    to: critic
    condition: quality_above_0_7         # registered in TransformRegistry

For CLI-runnable pipelines, load_workflows_file() (also behind serve --workflows-file, see Serve agents over MCP) expects the sequential schema instead: a top-level workflows: list where each entry has name, description, and steps with agent/input_map/condition — a one-to-one mirror of the WorkflowBuilder API:

workflows:
  - name: content_creation_pipeline
    description: Draft copy, review it, illustrate it
    steps:
      - agent: drafter
        input_map: {prompt: topic}
      - agent: critic
        input_map: {prompt: output}
      - agent: flux_pro
        input_map: {prompt: output}
        condition: "len(output) > 40"

4. Run and checkpoint from the CLI

Once a workflow is registered — via register_workflow() or a workflows file — run it by name. Each step's output is passed to the next, with input_map applied:

replicate-agent workflows run content_creation_pipeline --input '{"topic": "spring launch teaser"}'

╭─ Running Workflow ──────────────────────────────╮
│ Workflow: content_creation_pipeline  (3 steps)  │
│ Input: { "topic": "spring launch teaser" }      │
╰─────────────────────────────────────────────────╯

[Step 1/3] drafter
╭─ ✓ drafter  2841ms ─────────────────────────────╮
│ Introducing our spring collection — ...          │
╰─────────────────────────────────────────────────╯
[Step 2/3] critic
...

Useful flags: --json emits raw step chunks for piping, and --timeout caps each step at 300 seconds by default. For long pipelines, add --checkpoint-dir ./checkpoints — after every step the current state is written to <workflow>_step_<n>.json. If step 3 fails, fix the cause and resume with --resume-from 2 (the index is 0-based), skipping the steps that already succeeded:

replicate-agent workflows run content_creation_pipeline \
  --input '{"topic": "spring launch teaser"}' \
  --checkpoint-dir ./checkpoints --resume-from 2

[Step 3/3] flux_pro
╭─ ✓ flux_pro  8412ms ────────────────────────────╮
│ https://replicate.delivery/.../output.webp      │
╰─────────────────────────────────────────────────╯

Verification

Common mistakes

Next steps

• Routing — how the cost-aware router picks among tagged agents using learned cost and latency.
• Resilience & caching — circuit breakers, retries, and caches around every workflow step.
• API reference — full signatures for agent, AgentBuilder, WorkflowBuilder, and AgentWorkflow.