Reservo's pipeline, executed end to end

Description

The two previous lessons built each piece separately: what identifies a stage (02) and how data gets chained between two of them (03). This lesson brings them together over the module's complete case: the three stages of Reservo's real booking process — quote, validate the cancellation policy, confirm — running end to end with run_pipeline, with no change from lesson 03's. There's no new piece here — it's the first time you see the complete pipeline working, with each specialist's history printed and the final payload cited.

This is the module's centerpiece: the one that holds up lesson 05's cost comparison, lesson 06's guard, and lesson 08's mini-project's three scenarios.

Connection to the module

This lesson reuses, unchanged, PipelineStage (02), build_stage_task, extract_payload, and run_pipeline (03). The only new thing is the policy stage inserted in the middle of the sequence, and the three scripts (concept) that correspond to the complete case. Lesson 05 takes this lesson's executed result and adds the cost count compared against a supervisor.

Analogy: the assembly line, a full shift

The previous lessons showed one station of the line at a time: cutting, alone; cutting and welding, in a small two-station pipeline. This lesson is the complete line, on a real shift: the piece enters through the first station, passes through the second with no one having to step in, and comes out finished from the third — quoted, with its policy validated, and booked.

Worked example: the complete pipeline

import ast
import concurrent.futures
from dataclasses import dataclass
import reservo_tools as rt


def dispatch_parallel(tool_use_blocks, tools):
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(tool_use_blocks)) as pool:
        futures = [pool.submit(tools[b["name"]], **b["input"]) for b in tool_use_blocks]
        results = [f.result() for f in futures]
    return [
        {"type": "tool_result", "tool_use_id": b["id"], "content": str(r)}
        for b, r in zip(tool_use_blocks, results)
    ]


def run_agent_parallel(question, model_script, tools, max_iterations=10):
    messages = [{"role": "user", "content": question}]
    for step in range(max_iterations):
        turn = model_script[step]
        messages.append({"role": "assistant", "content": turn["content"]})
        if turn["stop_reason"] != "tool_use":
            return turn, messages
        tool_result_blocks = dispatch_parallel(turn["content"], tools)
        messages.append({"role": "user", "content": tool_result_blocks})
    raise RuntimeError(f"max_iterations reached ({max_iterations})")


POLICY_DOCS = {
    "no-show-policy": (
        "If a member doesn't show up for a confirmed booking and doesn't "
        "cancel at least 2 hours in advance, Reservo charges 50% of the "
        "quoted price as a no-show fee."
    ),
    "cancellation-policy": (
        "Bookings can be cancelled at no charge up to 2 hours before the "
        "booked time slot. Cancellations within those 2 hours incur the "
        "no-show fee."
    ),
}


def search_docs(query):
    q = query.lower()
    if "cancel" in q:
        return f"[cancellation-policy] {POLICY_DOCS['cancellation-policy']}"
    if "show" in q and "n't" in q:
        return f"[no-show-policy] {POLICY_DOCS['no-show-policy']}"
    return "No relevant policy was found for that question."


SPECIALISTS = {
    "booking_agent": {
        "tools": {
            "list_rooms": rt.list_rooms, "get_quote": rt.get_quote,
            "book_room": rt.book_room, "cancel_booking": rt.cancel_booking,
        },
        "expertise": "quote, book, and cancel rooms",
    },
    "policy_agent": {
        "tools": {"search_docs": search_docs},
        "expertise": "answer policy questions (cancellation, no-show)",
    },
}


def run_specialist(name, task, model_script):
    tools = SPECIALISTS[name]["tools"]
    return run_agent_parallel(task, model_script, tools)


@dataclass
class PipelineStage:
    kind: str
    name: str
    label: str


def last_tool_result(history):
    for m in reversed(history):
        if isinstance(m["content"], list):
            for b in m["content"]:
                if b["type"] == "tool_result":
                    try:
                        return ast.literal_eval(b["content"])
                    except (ValueError, SyntaxError):
                        return b["content"]
    return None


def build_stage_task(stage, payload):
    if stage.kind == "quote":
        return f"Quote {payload['room']} {payload['tier']} {payload['hours']}h."
    if stage.kind == "validate_policy":
        return (f"What's the cancellation policy for a {payload['hours']}h "
                 f"{payload['room']} booking, before confirming it?")
    if stage.kind == "confirm":
        if payload.get("cleared_to_book"):
            return (f"Book {payload['room']} {payload['tier']} {payload['hours']}h "
                     f"for {payload['member']} -- the cancellation policy has "
                     f"already been validated.")
        return (f"Book {payload['room']} {payload['tier']} {payload['hours']}h "
                 f"for {payload['member']}.")
    raise ValueError(f"don't know how to build the task for stage {stage.kind!r}")


def extract_payload(stage, history, payload):
    new_payload = dict(payload)
    result = last_tool_result(history)
    if stage.kind == "quote":
        new_payload["price_cents"] = result["price_cents"]
    elif stage.kind == "validate_policy":
        new_payload["cancellation_policy"] = result
        new_payload["cleared_to_book"] = True
    elif stage.kind == "confirm":
        new_payload["booking_id"] = result["booking_id"]
        new_payload["confirmed"] = result["confirmed"]
    return new_payload


def run_pipeline(stages, model_scripts, initial_payload):
    payload = dict(initial_payload)
    trace = []
    for i, stage in enumerate(stages):
        task = build_stage_task(stage, payload)
        final, history = run_specialist(stage.name, task, model_scripts[i])
        payload = extract_payload(stage, history, payload)
        trace.append({
            "stage": i + 1, "kind": stage.kind, "agent": stage.name,
            "label": stage.label, "task": task, "history": history,
            "output": final["content"][0]["text"],
        })
    return payload, trace


# Reservo's complete pipeline: quote -> validate -> confirm.
PIPELINE_STAGES = [
    PipelineStage(kind="quote", name="booking_agent", label="quote"),
    PipelineStage(kind="validate_policy", name="policy_agent",
                  label="validate the cancellation policy"),
    PipelineStage(kind="confirm", name="booking_agent", label="confirm the booking"),
]

INITIAL_PAYLOAD = {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}

# The three scripts (concept, claude-sonnet-5) -- one per stage.
model_script_quote = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Focus pro 3h costs 6000 cents."}]},
]

model_script_policy = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "search_docs",
         "input": {"query": "cancellation policy"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": (
            "You can cancel at no charge up to 2 hours before the booked "
            "time. There's no impediment to confirming."
        )}]},
]

model_script_confirm = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "I booked Focus pro 3h for Ana (confirmation #1)."}]},
]

model_scripts = [model_script_quote, model_script_policy, model_script_confirm]

payload, trace = run_pipeline(PIPELINE_STAGES, model_scripts, INITIAL_PAYLOAD)

for step in trace:
    print(f"--- stage {step['stage']} ({step['label']}) -> {step['agent']} ---")
    print(f"task: {step['task']!r}")
    for i, m in enumerate(step["history"]):
        role, content = m["role"], m["content"]
        if isinstance(content, str):
            print(f"  [{i}] {role:<9} question: {content!r}")
            continue
        for block in content:
            if block["type"] == "tool_use":
                print(f"  [{i}] {role:<9} tool_use({block['name']}): {block['input']}")
            elif block["type"] == "tool_result":
                print(f"  [{i}] {role:<9} tool_result: {block['content']}")
            elif block["type"] == "text":
                print(f"  [{i}] {role:<9} final text: {block['text']!r}")
    print()

print("final payload:", payload)

What to expect (over a fresh, disposable Reservo instance):

--- stage 1 (quote) -> booking_agent ---
task: 'Quote Focus pro 3h.'
  [0] user      question: 'Quote Focus pro 3h.'
  [1] assistant tool_use(get_quote): {'room': 'Focus', 'tier': 'pro', 'hours': 3}
  [2] user      tool_result: {'price_cents': 6000}
  [3] assistant final text: 'Focus pro 3h costs 6000 cents.'

--- stage 2 (validate the cancellation policy) -> policy_agent ---
task: "What's the cancellation policy for a 3h Focus booking, before confirming it?"
  [0] user      question: "What's the cancellation policy for a 3h Focus booking, before confirming it?"
  [1] assistant tool_use(search_docs): {'query': 'cancellation policy'}
  [2] user      tool_result: [cancellation-policy] Bookings can be cancelled at no charge up to 2 hours before the booked time slot. Cancellations within those 2 hours incur the no-show fee.
  [3] assistant final text: "You can cancel at no charge up to 2 hours before the booked time. There's no impediment to confirming."

--- stage 3 (confirm the booking) -> booking_agent ---
task: 'Book Focus pro 3h for Ana -- the cancellation policy has already been validated.'
  [0] user      question: 'Book Focus pro 3h for Ana -- the cancellation policy has already been validated.'
  [1] assistant tool_use(book_room): {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'member': 'Ana'}
  [2] user      tool_result: {'booking_id': 1, 'confirmed': True}
  [3] assistant final text: 'I booked Focus pro 3h for Ana (confirmation #1).'

final payload: {'room': 'Focus', 'tier': 'pro', 'hours': 3, 'member': 'Ana', 'price_cents': 6000, 'cancellation_policy': '[cancellation-policy] Bookings can be cancelled at no charge up to 2 hours before the booked time slot. Cancellations within those 2 hours incur the no-show fee.', 'cleared_to_book': True, 'booking_id': 1, 'confirmed': True}

Three stages, three specialists invoked (two of them the same one, booking_agent, at different points), and no point where anyone had to decide "which one's next?". Notice stage 3's task: 'Book Focus pro 3h for Ana -- the cancellation policy has already been validated.' — that sentence built itself, because payload['cleared_to_book'] was already True when build_stage_task built it. No member and no model had to repeat that information; it traveled, real, from stage 2.

Composing the final answer (concept)

The pipeline ended with a complete payload, but the member doesn't need to see a Python dictionary — they need a prose answer. Just like Module 2's supervisor had a Step 4 (aggregate), the pipeline needs a final synthesis step:

# Final step (concept, claude-sonnet-5): composes the answer for the member
# from the complete payload -- no new data, just wording.
final_response = (
    f"The quote for {payload['room']} {payload['tier']} {payload['hours']}h "
    f"is {payload['price_cents']} cents. I checked the cancellation "
    f"policy: you can cancel at no charge up to 2 hours in advance. I "
    f"booked the room for {payload['member']} (confirmation #{payload['booking_id']})."
)
print("--- composed answer for the member (concept) ---")
print(final_response)

What to expect:

--- composed answer for the member (concept) ---
The quote for Focus pro 3h is 6000 cents. I checked the cancellation policy: you can cancel at no charge up to 2 hours in advance. I booked the room for Ana (confirmation #1).

This answer combines information from all three stages — the price from 1, the policy from 2, the confirmation from 3 — into a single message. Just like COMPOSE_CALLS in Module 2, this synthesis is concept: one more model turn, which lesson 05 counts as part of the total coordination cost.

Common mistakes

1. Thinking this pipeline handles any booking task. It only handles the exact sequence PIPELINE_STAGES defines — a task that, say, only asked to quote without booking doesn't fit this pipeline as it is (Module 2 already solves that simpler case with a single specialist).

2. Running this lesson after another guide example in the same interpreter. If book_room doesn't return booking_id: 1, the process already had bookings from an earlier run. Every lesson in this guide assumes its own fresh, disposable Reservo instance.

3. Confusing "three stages" with "three model calls." Each stage alone costs two model calls (a tool_use turn, a final-text turn) — lesson 05 does the complete accounting, including the coordination cost surrounding all three.

4. Forgetting cleared_to_book comes from stage 2, not from an assumption. build_stage_task's if payload.get("cleared_to_book") depends on stage 2 having run BEFORE — the pipeline's fixed order guarantees that — but if you ever reordered the stages without thinking it through, cleared_to_book wouldn't exist yet when stage 3 tried to read it.

5. Thinking the final composed answer "does nothing new." Unlike Module 2, lesson 02 — where the aggregated answer turned out to be, by coincidence, the same text the specialist already had — here the synthesis DOES combine three distinct pieces of information into a message no individual stage had in full.

Exercises

Exercise 1: Confirm your own run (Easy)

Run this lesson's complete pipeline yourself and confirm, line by line, that your output matches the "What to expect" block above. Pay special attention to the booking_id and the cleared_to_book in the final payload.

Exercise 2: Run the same pipeline with Boardroom pro 4h (Medium)

Repeat the complete pipeline, in the SAME session where you already ran the worked example, with INITIAL_PAYLOAD = {"room": "Boardroom", "tier": "pro", "hours": 4, "member": "Sofía"} and its corresponding three scripts. Confirm the price (8000 * 4 * 80 // 100) and the booking_id.

INITIAL_PAYLOAD_BR = {"room": "Boardroom", "tier": "pro", "hours": 4, "member": "Sofía"}
script_quote_br = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 4}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Boardroom pro 4h costs 25600 cents."}]},
]
script_confirm_br = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 4, "member": "Sofía"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "I booked Boardroom pro 4h for Sofía (confirmation #2)."}]},
]
payload_br, trace_br = run_pipeline(
    PIPELINE_STAGES, [script_quote_br, model_script_policy, script_confirm_br], INITIAL_PAYLOAD_BR,
)
print("by-hand calculated price:", 8000 * 4 * 80 // 100)
print("final payload:", payload_br)

by-hand calculated price: 25600
final payload: {'room': 'Boardroom', 'tier': 'pro', 'hours': 4, 'member': 'Sofía', 'price_cents': 25600, 'cancellation_policy': '[cancellation-policy] Bookings can be cancelled at no charge up to 2 hours before the booked time slot. Cancellations within those 2 hours incur the no-show fee.', 'cleared_to_book': True, 'booking_id': 2, 'confirmed': True}

Exercise 3: What happens if stage 2 gets skipped? (Hard)

Build an alternative PIPELINE_STAGES with only two stages — quote and confirm, without validate_policy — and run it with INITIAL_PAYLOAD. Confirm the confirm stage's task text changes compared to the worked example, and explain why.

STAGES_NO_POLICY = [
    PipelineStage(kind="quote", name="booking_agent", label="quote"),
    PipelineStage(kind="confirm", name="booking_agent", label="confirm the booking"),
]
payload_np, trace_np = run_pipeline(
    STAGES_NO_POLICY, [model_script_quote, model_script_confirm], INITIAL_PAYLOAD,
)
print("confirm stage's task:", trace_np[1]["task"])

confirm stage's task: 'Book Focus pro 3h for Ana.'

Summary and next step

• We ran Reservo's complete pipeline: quote → validate the cancellation policy → confirm, with run_pipeline unchanged from lesson 03.
• Stage 1's price and stage 2's validation traveled, real, all the way to stage 3 — the confirmation task built itself, mentioning the already-validated policy, without anyone repeating that information.
• We composed a final answer (concept) that combines all three stages into a single message for the member — the first time in this module where the final synthesis combines more than one source of real information.
• At no point across the three stages did anyone need to decide "which one's next?" — the order was fixed in PIPELINE_STAGES before this specific request ever existed.

Next lesson: 05 — Measuring a pipeline's coordination cost. We count, with real numbers, how much this pipeline saves over a supervisor that had to decide at each of the three stages.

Additional resources

1. Anthropic — Building effective agents — The "prompt chaining" pattern with verification gates between steps, run end to end in this lesson with Reservo.
2. Anthropic — Multi-agent research system — A flow with mandatory stages where one's result feeds directly into the next, with no intermediate decision point.
3. Anthropic — Messages API reference — The exact shape of tool_use/tool_result/stop_reason every stage of this pipeline respects, unchanged.
4. Python — dataclasses — The module behind PipelineStage, reused unchanged from lesson 02.