Build a Multi-Agent Workflow

This page shows how to build a simple multi-agent workflow using CAIP Agents SDK and LangGraph.

The workflow contains:

• A router node that decides where the request should go
• A billing worker for invoice or payment questions
• A technical worker for diagnostics and support issues
• A general worker for everything else
• Optional human approval for sensitive actions
• Domain-specific tools attached only to the workers that need them

Workflow Overview

User message
    |
    v
Router
    |-- billing   --> Billing worker    --> Billing tools    --> Billing worker --> END
    |-- technical --> Technical worker  --> Technical tools  --> Technical worker --> END
    |-- general   --> General worker                                        --> END
    |-- escalate  --> HITL approval gate --> Escalation worker              --> END

This pattern keeps the design easy to understand. The router only decides the route. Each worker owns one domain. Tools are limited to the workers that need them.

1. Create a LangGraph Agent

Create the SDK client and select LangGraph as the framework.

The important part is graph_factory=build_support_graph. This tells the SDK to use your graph instead of the default single-agent graph.

from caip_agents_sdk import CAIPAgentsClient

client = CAIPAgentsClient()

agent = client.create_agent(
    framework="langgraph",
    agent_id="your-agent-id",
    graph_factory=build_support_graph,
)

2. Register Tools Before Initialization

Tools are plain Python functions. Register them before await agent.initialize() so the graph factory receives the full tool list.

@agent.tool_plain
def check_invoice(customer_id: str) -> str:
    """Check invoice status for a customer."""
    return f"Invoice for {customer_id}: paid"


@agent.tool_plain
def run_diagnostics(customer_id: str, symptom: str) -> str:
    """Run technical diagnostics for a customer issue."""
    return f"Diagnostics complete for {customer_id}: probable cause is {symptom}"


@agent.tool_plain
def create_escalation_ticket(customer_id: str, reason: str) -> str:
    """Create an escalation ticket for a customer."""
    return f"Escalation ticket created for {customer_id}. Reason: {reason}"

tip

Use clear tool names and docstrings. The model uses these descriptions to decide when a tool is useful.

3. Build the Graph Factory

The graph factory receives two important objects from the SDK:

• llm: the CAIP-configured LLM client
• tools: the tools you registered on the agent

The graph factory returns a compiled LangGraph workflow.

from typing import Literal

from langgraph.types import interrupt

from caip_agents_sdk import (
    Command,
    END,
    MessagesState,
    START,
    StateGraph,
    SystemMessage,
    ToolNode,
    tools_condition,
)


class SupportState(MessagesState):
    route: str


ROUTER_PROMPT = """
Classify the latest user request into one label only:
- billing
- technical
- general
- escalate
Respond with a single word only.
""".strip()


def extract_text(content) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "".join(
            item.get("text", "")
            for item in content
            if isinstance(item, dict) and item.get("type") == "text"
        )
    return ""


def build_support_graph(llm, tools, checkpointer=None):
    tool_map = {tool.name: tool for tool in tools}

    billing_tools = [tool_map["check_invoice"]]
    technical_tools = [tool_map["run_diagnostics"]]
    escalation_tools = [tool_map["create_escalation_ticket"]]

    billing_llm = llm.bind_tools(billing_tools)
    technical_llm = llm.bind_tools(technical_tools)
    escalation_llm = llm.bind_tools(escalation_tools)

    async def router(state: SupportState) -> Command[
        Literal[
            "billing_worker",
            "technical_worker",
            "general_worker",
            "approval_gate",
        ]
    ]:
        messages = [SystemMessage(content=ROUTER_PROMPT)] + list(state["messages"])
        response = await llm.ainvoke(messages)
        route = extract_text(response.content).strip().lower()

        if route not in {"billing", "technical", "general", "escalate"}:
            route = "general"

        if route == "escalate":
            return Command(goto="approval_gate", update={"route": route})

        return Command(goto=f"{route}_worker", update={"route": route})

    def make_worker(worker_llm, prompt: str):
        async def worker(state: SupportState):
            messages = [SystemMessage(content=prompt)] + list(state["messages"])
            response = await worker_llm.ainvoke(messages)
            return {"messages": [response]}

        return worker

    def approval_gate(state: SupportState) -> Command[
        Literal["escalation_worker", "general_worker"]
    ]:
        decision = interrupt(
            {
                "type": "escalation_approval",
                "reason": "The workflow classified this request as escalation.",
                "options": ["approve", "decline"],
            }
        )

        if str(decision).strip().lower() == "approve":
            return Command(goto="escalation_worker")

        return Command(goto="general_worker", update={"route": "general"})

    builder = StateGraph(SupportState)
    builder.add_node("router", router)
    builder.add_node("billing_worker", make_worker(billing_llm, "You are a billing specialist."))
    builder.add_node("billing_tools", ToolNode(billing_tools))
    builder.add_node("technical_worker", make_worker(technical_llm, "You are a technical specialist."))
    builder.add_node("technical_tools", ToolNode(technical_tools))
    builder.add_node("general_worker", make_worker(llm, "You are a general support assistant."))
    builder.add_node("approval_gate", approval_gate)
    builder.add_node("escalation_worker", make_worker(escalation_llm, "You are an escalation specialist."))
    builder.add_node("escalation_tools", ToolNode(escalation_tools))

    builder.add_edge(START, "router")

    builder.add_conditional_edges(
        "billing_worker",
        tools_condition,
        {"tools": "billing_tools", "__end__": END},
    )
    builder.add_edge("billing_tools", "billing_worker")

    builder.add_conditional_edges(
        "technical_worker",
        tools_condition,
        {"tools": "technical_tools", "__end__": END},
    )
    builder.add_edge("technical_tools", "technical_worker")

    builder.add_edge("general_worker", END)
    builder.add_edge("approval_gate", "escalation_worker")
    builder.add_edge("approval_gate", "general_worker")
    builder.add_conditional_edges(
        "escalation_worker",
        tools_condition,
        {"tools": "escalation_tools", "__end__": END},
    )
    builder.add_edge("escalation_tools", "escalation_worker")

    if checkpointer is not None:
        return builder.compile(checkpointer=checkpointer)
    return builder.compile()

4. Understand Conditional Routing

Conditional routing decides which node runs next based on runtime state or model output.

In this example, routing happens in the router node:

if route == "escalate":
    return Command(goto="approval_gate", update={"route": route})

return Command(goto=f"{route}_worker", update={"route": route})

This means:

• billing goes to billing_worker
• technical goes to technical_worker
• general goes to general_worker
• escalate goes to approval_gate

The update={"route": route} part stores the route in graph state. This is useful for debugging and evals.

5. Add Nodes and Edges

Use add_node() to register each workflow step.

builder.add_node("router", router)
builder.add_node("billing_worker", billing_worker)
builder.add_node("billing_tools", ToolNode(billing_tools))

Use add_edge() when the next step is fixed.

builder.add_edge(START, "router")
builder.add_edge("billing_tools", "billing_worker")
builder.add_edge("general_worker", END)

Use add_conditional_edges() when the next step depends on the result of a node.

builder.add_conditional_edges(
    "billing_worker",
    tools_condition,
    {"tools": "billing_tools", "__end__": END},
)

This tells LangGraph:

• If the worker asks for a tool, go to billing_tools
• If no tool is needed, end the workflow

tip

Start with a small graph first. Add one router, two workers, and one tool loop. Once that works, add more workers and approval paths.

6. Initialize and Run

import asyncio
import os

from dotenv import load_dotenv


async def main() -> None:
    load_dotenv()

    await agent.initialize()

    thread = await client.create_thread(
        agent_id=os.getenv("CAIP_AGENT_ID"),
        thread_data={"title": "LangGraph Support Workflow", "status": "open"},
    )
    agent.thread_id = thread.threadId

    result = await agent.run("Customer C001 has an invoice question.")
    print(result.output)


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

7. Add Checkpointing When Needed

Use checkpointing when you need durable state, approval flows, or state inspection. HITL requires checkpointing because the graph must pause and later resume from the same state.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

agent = client.create_agent(
    framework="langgraph",
    agent_id="your-agent-id",
    graph_factory=build_support_graph,
    checkpointer=checkpointer,
)

When checkpointing is enabled, set agent.thread_id before calling run(), resume(), or get_state().

8. Use Human-in-the-Loop (HITL)

Human-in-the-loop is useful when a workflow should not continue without human approval.

Common examples:

• Refunds above a threshold
• Account closure
• Legal or compliance-sensitive answers
• Escalation to human support

In the graph, use interrupt() inside a node:

def approval_gate(state: SupportState):
    decision = interrupt(
        {
            "type": "escalation_approval",
            "reason": "The workflow classified this request as escalation.",
            "options": ["approve", "decline"],
        }
    )

    if str(decision).strip().lower() == "approve":
        return Command(goto="escalation_worker")

    return Command(goto="general_worker")

When the graph pauses, agent.run() returns an interrupted result:

result = await agent.run("Customer C001 wants to escalate this issue.")

if result.is_interrupted():
    print(result.interrupts())

After a human decision, resume the graph:

resumed = await agent.resume("approve")
print(resumed.output)

important

For HITL, always use a checkpointer and set agent.thread_id before running the workflow.

9. Perform Evals

Evals help you check whether the workflow routes requests correctly and returns useful responses.

Start with simple functional evals before building advanced scoring.

Route Accuracy Eval

Create a small set of test cases with expected routes:

eval_cases = [
    ("Customer C001 has an invoice question", "billing"),
    ("Customer C002 internet is not working", "technical"),
    ("What are your support hours?", "general"),
    ("I want to speak to a manager now", "escalate"),
]

Run each case and inspect the route saved in the graph state:

async def run_route_evals(agent):
    passed = 0

    for query, expected_route in eval_cases:
        result = await agent.run(query)
        actual_route = result.raw.get("route")

        if actual_route == expected_route:
            passed += 1
            print(f"PASS: {query} -> {actual_route}")
        else:
            print(f"FAIL: {query} -> {actual_route}, expected {expected_route}")

    print(f"Route accuracy: {passed}/{len(eval_cases)}")

HITL Eval

For HITL, verify that sensitive cases pause instead of completing automatically:

result = await agent.run("I want to speak to a manager now")

if result.is_interrupted():
    print("PASS: workflow paused for human approval")
else:
    print("FAIL: workflow should have paused")

Response Quality Eval

For response quality, start with simple checks:

• Does the response answer the user request?
• Did the workflow use the right tool?
• Did the workflow avoid sensitive actions without approval?
• Is the final answer clear and concise?

As your workflow grows, you can add automated scoring, human review, or regression tests around common scenarios.

10. Design Checklist

Before building your own workflow, define:

• Which domains need specialist workers
• Which tools each worker can use
• Which routes the router can return
• Which actions need human approval
• What state must be preserved
• What success criteria you will measure

This keeps the graph understandable and easier to operate in production.

11. Full Example

For a complete customer support implementation based on the SDK repository example, see LangGraph Customer Support.