Tutorial 5: a refund agent with approval and a safe payment

In this tutorial, you build an agent that handles refund claims. A citizen writes a complaint. The model reads it and proposes a refund. An employee approves the refund, and the agent pays it. This agent is not a ReAct loop: you write its graph yourself, node by node.

The payment is an action in the world: it moves money. The agent must never pay twice, also when the connection fails in the middle of the payment. At the end of the tutorial, you make the connection fail on purpose and see that the agent pays once.

Do tutorial 3 first: this tutorial uses the prompt file, and the difference between the agent and the graph, from there.

Set upCopy Anchor Link

Use the repository and the environment variables of tutorial 1. Make a folder for this tutorial:

mkdir -p my_tutorials/05-refund-graph

Step 1: write the prompt fileCopy Anchor Link

Make my_tutorials/05-refund-graph/prompt.yaml:

system: |
  You handle refund requests for the town of Springfield.
  An employee approves each refund before it is paid, so never promise a refund.
  If the order number or the amount is missing, ask the citizen for it.

Step 2: write the proposal toolCopy Anchor Link

Make my_tutorials/05-refund-graph/agent.py. These are all its imports, and the lines that load and check the prompt file:

import asyncio
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Annotated

from openai import AsyncOpenAI
from pydantic import BaseModel, Field

from soev_agents.adapters import InMemoryTraceStore, OpenAIChatModel
from soev_agents.domain import EffectIntent, EffectResult, Event, Input, ModelOutput, ToolOutput, Trace
from soev_agents.runtime import (
    Agent,
    AgentCatalog,
    EffectCancellation,
    Effector,
    EffectRecovery,
    Graph,
    LlmStep,
    RequestStep,
    ToolStep,
    abandon,
    open_requests,
    tool,
    tool_def,
)
from soev_agents.toolkit import MessageBuilder, PromptFile, Text, rendered_by

PROMPTS = PromptFile.load(Path(__file__).with_name("prompt.yaml"))
PROMPTS.check_variables(set())

This agent renders the prompt itself (step 6), and gives the template no variables. check_variables states that in the code: it lists the variables that the code gives, here none. If the prompt file reads a variable, the program fails when it starts, not during a conversation.

The model does not pay. It only proposes a refund, with a tool:

class Claim(BaseModel):
    order: str
    amount: float
    reason: str


@tool(
    name="propose_refund",
    description="Propose a refund for an employee to approve. Use it when a citizen asks for a refund and gives the "
    "order number and the amount.",
)
async def propose_refund(
    order: Annotated[str, Field(pattern=r"^P-\d{4}$", description="The order number, such as P-2041.")],
    amount: Annotated[float, Field(gt=0, le=500, description="The amount to refund, in euros.")],
    reason: Annotated[str, Field(description="Why the citizen gets the refund, in one sentence.")],
) -> ToolOutput:
    claim = Claim(order=order, amount=amount, reason=reason)
    return ToolOutput(text="The refund waits for the approval of an employee.", data=claim.model_dump())

The Field rules check each argument before the tool runs. An order number without P-, or an amount above 500 euros, gets an error, and the model can try again. The tool records the claim in data, where the next nodes read it.

Step 3: write the payment serviceCopy Anchor Link

The payment service is the outside world. In a real agent, it is the API of a bank:

@dataclass
class Payments:
    """A payment service: it pays each key at most once. In a real agent, this is the API of a bank."""

    paid: dict[str, float] = field(default_factory=dict)
    lose_connection: bool = False

    async def pay(self, key: str, amount: float) -> None:
        self.paid.setdefault(key, amount)
        if self.lose_connection:
            self.lose_connection = False
            raise ConnectionError("the connection was lost after the payment")

    async def find(self, key: str) -> float | None:
        return self.paid.get(key)


def payment_key(claim: Claim) -> str:
    return f"refund-{claim.order}"

Each payment has a key. The service pays each key at most once, as many bank APIs do with an idempotency key: a second payment with the same key changes nothing. One order gets one refund, so the key comes from the order number.

lose_connection makes the next payment fail after the money moved. The agent does not know if the payment happened. This is the hard case.

Payments is not frozen, because it is the bank in this example: its state is the state of the outside world, not of the agent. The agent keeps no state of its own.

Step 4: write the payment as an effectCopy Anchor Link

A plain step can run again after a failure. For a payment, that is not safe. An effect step, an EffectStep, is a node in three parts:

1. intend writes down what the agent is about to do, as an EffectIntent. The framework records the intent before anything happens.
2. The effector, an Effector, does the action and returns an EffectResult.
3. recovery, an EffectRecovery, runs when the trace holds an intent without a result. It decides what happened.

The effector pays the claim:

@dataclass(frozen=True)
class PayRefund:
    payments: Payments

    async def perform(self, intent: EffectIntent) -> EffectResult:
        claim = Claim.model_validate(intent.args)
        await self.payments.pay(payment_key(claim), claim.amount)
        return EffectResult(actor="system", data={"payment": payment_key(claim)})

The recovery asks the payment service what happened. If the service has the payment, the recovery records the result. If not, it pays:

@dataclass(frozen=True)
class LookUpPayment:
    payments: Payments

    async def __call__(self, trace: Trace, intent: EffectIntent, effector: Effector) -> list[Event]:
        claim = Claim.model_validate(intent.args)
        if await self.payments.find(payment_key(claim)) is not None:
            return [EffectResult(actor="system", data={"payment": payment_key(claim)})]
        return [await effector.perform(intent)]

A recovery has three choices: look up the outcome, as here, do the action again when that is safe, or ask a person. The framework never repeats an action without the recovery.

The effect step puts the three parts together:

@dataclass(frozen=True)
class Refund:
    effector: Effector
    recovery: EffectRecovery
    cancellation: EffectCancellation = abandon

    def intend(self, trace: Trace) -> EffectIntent:
        proposal = trace.latest(ToolOutput)
        assert proposal is not None
        return EffectIntent(kind="refund", args=Claim.model_validate(proposal.data).model_dump())

intend reads the claim from the trace, from the result of the proposal tool. cancellation says what to do when someone cancels the turn during the payment. abandon leaves an open payment to the recovery.

Refund is an effect step because it has the four attributes of EffectStep: effector, recovery, cancellation and intend. The runner checks for all four. A node without one of them is a plain step.

Step 5: write the routesCopy Anchor Link

A graph goes from node to node along edges. A router chooses the next node: a function that reads the trace and returns the name of the next node, or None to stop.

def after_intake(trace: Trace) -> str | None:
    answer = trace.latest(ModelOutput)
    return "propose" if answer is not None and answer.calls_tools else None


def after_proposal(trace: Trace) -> str:
    proposal = trace.latest(ToolOutput)
    return "approve" if proposal is not None and proposal.error is None else "intake"


def after_approval(trace: Trace) -> str:
    decision = trace.last()
    return "refund" if isinstance(decision, Input) and decision.payload is True else "decline"


def after_refund(trace: Trace) -> str:
    result = trace.latest(EffectResult)
    return "confirm" if result is not None and result.error is None else "decline"

• After intake: if the model called the tool, go to propose. If the model answered in text, for example to ask for the order number, stop. The citizen answers in a new turn.
• After propose: if the tool refused the arguments, go back to the model, which reads the error.
• After approve: the employee's answer is the last event. True goes to the payment.
• After refund: a result without an error goes to the confirmation.

The last two nodes write the answer to the citizen. They are plain steps: async functions from the trace to new events. They need no model:

async def confirm(trace: Trace) -> list[Event]:
    intent = trace.latest(EffectIntent)
    assert intent is not None
    claim = Claim.model_validate(intent.args)
    return [ModelOutput(actor="system", content=f"We refunded {claim.amount:.2f} euros for order {claim.order}.")]


async def decline(trace: Trace) -> list[Event]:
    return [ModelOutput(actor="system", content="We cannot refund this order. An employee will contact you.")]

Step 6: write the agent and its graphCopy Anchor Link

In tutorials 1 to 3, ReActAgent was the agent, and it built a ReAct graph for each turn. This agent needs a graph of its own, so you write the agent class yourself. An agent is a class with two parts:

• input: the class that each input that opens a turn must parse into. Text is plain text. An answer to a request is checked against the expects of that request instead.
• build: a method that makes the graph for one turn.

class RefundDesk(Agent[Payments, Text]):
    input = Text

    def __init__(self, model: OpenAIChatModel) -> None:
        self.model = model

    def build(self, resources: Payments, turn: Text, model: str | None) -> Graph:
        prompts = PROMPTS.for_model(model)
        intake = MessageBuilder(
            render_input=rendered_by(Text),
            model=self.model.resolve(model),
            prompt=lambda citation: prompts.system(),
            tools=(tool_def(propose_refund),),
        )
        graph = Graph()
        graph.add_node("intake", LlmStep(builder=intake, model=self.model))
        graph.add_node("propose", ToolStep(tools={"propose_refund": propose_refund}))
        graph.add_node("approve", RequestStep(kind="approval", actor="system", expects={"type": "boolean"}))
        graph.add_node("refund", Refund(effector=PayRefund(resources), recovery=LookUpPayment(resources)))
        graph.add_node("confirm", confirm)
        graph.add_node("decline", decline)
        graph.set_entry_point("intake")
        graph.add_conditional_edge("intake", after_intake)
        graph.add_conditional_edge("propose", after_proposal)
        graph.add_conditional_edge("approve", after_approval)
        graph.add_conditional_edge("refund", after_refund)
        return graph

The class line, Agent[Payments, Text], gives the types of the two things that build gets from outside:

• resources, a Payments: the adapters of this request, as in tutorial 3. build gives them to the effect step.
• turn, a Text: the input of the turn, parsed.

The third argument, model, is the model that the input names. PROMPTS.for_model selects the texts of the prompt file for that model.

ReActAgent builds its graph from the same parts that you use here:

• MessageBuilder makes the request to the model from the trace: the system prompt, the messages and the offered tools. rendered_by(Text) shows each input as a user message. tool_def(propose_refund) is what the model sees of the tool: its name, its description and the schema of its arguments. self.model.resolve(model) gives the name of the model that the call goes to: the model of the input, or the default of the adapter. prompt is a function of an example citation marker. This agent does not cite, so it ignores the marker.
• LlmStep calls the model with that request and records its answer.
• ToolStep runs the tool calls of the last answer.
• RequestStep stops the run and asks for outside input. expects is a JSON Schema for the answer: here, true or false. While the request is open, the thread waits.
• add_conditional_edge connects a node to its router. confirm and decline have no edge, so the turn ends after them.

Why the agent and the graph are separateCopy Anchor Link

In tutorial 3, you saw the first reason: the agent is shared by all users, and each turn gets a graph with the resources of its own request. There are three more reasons.

Some frameworks, such as LangGraph, make the graph once, when the program starts. Then no request exists yet: no user, no store, no turn. So each step must get those values as arguments each time that it runs. soev-agents makes the graph at the start of each turn instead, with the values of that turn in it. This gives:

1. A bad turn is refused before the framework records it. build runs before the framework records the input. If the turn cannot run, for example because the input names a model that the agent does not know, the client gets a refusal and the thread does not change. With a graph that exists from the start, the framework finds the problem only inside a step, after it recorded the input. The thread then holds a turn that cannot run.
2. Steps and tools get only the trace. A step, such as confirm, is a function of the trace. The effect step already holds Payments, because build gave it. A tool gets only its arguments, because bind gave it the rest. That is why the tests of tutorial 2 call a tool directly, with no framework.
3. The graph can differ for each turn. build can choose the nodes from the turn and the model. For example, it can leave out the tools for a model that cannot call tools.

A graph needs nothing that the trace does not have. After a crash, the framework builds the graph again from the input that the trace recorded (step 7, resuming=True).

Step 7: run the refundCopy Anchor Link

async def main() -> None:
    payments = Payments(lose_connection=True)

    async def resources(agent: Agent[Payments, Text]) -> Payments:
        return payments

    model = OpenAIChatModel(AsyncOpenAI(), model=os.environ["OPENAI_MODEL"])
    agents = AgentCatalog(
        InMemoryTraceStore(), {"refunds": RefundDesk(model)}, version="tutorial-5", resources=resources
    )

    complaint = "I paid twice for my parking permit, order P-2041. Please refund the 120 euros."
    thread, _ = await agents.open("refunds", Input(payload=complaint, model=model.model))
    trace = await thread.resume()
    [request] = open_requests(trace)
    proposal = trace.latest(ToolOutput)
    assert proposal is not None
    print("Waiting for approval:", Claim.model_validate(proposal.data))

Payments(lose_connection=True) makes the first payment fail after the money moved. The run stops at the approve node. open_requests returns the open request.

Now the employee approves. The answer is an Input. send gets the request that it answers as answering:

    thread = await agents.thread(thread.id, resuming=True)
    trace = await thread.send(Input(payload=True, actor="user"), answering=request)
    print("After the approval:", (await thread.status()).state, type(trace.last()).__name__)

resuming=True builds the graph for the turn in progress. The framework checks the answer against expects before it records it: an answer such as "yes" is refused.

The payment fails, so the thread stops with a Failure. Resume it:

    trace = await thread.resume()
    print("After the resume:", (await thread.status()).state)
    print("Answer:", trace.by_type(ModelOutput)[-1].content)
    print("Payments:", payments.paid)

    for event in trace.by_type(Event):
        print(event.seq, type(event).__name__, event.origin or "")


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

Run the agent:

uv run python my_tutorials/05-refund-graph/agent.py

The output is similar to this. The reason is in your model's words:

Waiting for approval: order='P-2041' amount=120.0 reason='The citizen paid twice for a parking permit.'
After the approval: failed Failure
After the resume: finished
Answer: We refunded 120.00 euros for order P-2041.
Payments: {'refund-P-2041': 120.0}
0 Opened
1 Input
2 ModelOutput intake
3 ToolOutput propose
4 Request approve
5 Input
6 EffectIntent refund
7 Failure refund
8 EffectResult refund
9 ModelOutput confirm
10 Finished

Read the trace from event 4:

• 4 Request: the agent asked for approval, as a Request, and the thread waited.
• 5 Input: the employee approved.
• 6 EffectIntent: the agent recorded what it was about to do, before the payment.
• 7 Failure: the payment raised an error after the money moved.
• 8 EffectResult: on resume, the trace held an intent without a result. The framework did not pay again. It ran the recovery, which found the payment.
• 9 ModelOutput: the confirmation for the citizen, a ModelOutput that no model wrote.

Payments holds one payment. In this example, the key also protects the payment: a second payment with the same key changes nothing. But many services have no idempotency key, and for some actions there is no outcome to look up. A plain step then pays twice. With an effect step, the recovery decides what happens, so the agent never guesses.

The same happens when the process dies between event 6 and event 8, for example in a deploy. The trace is in the store, and the next process that resumes the thread runs the recovery.

What you builtCopy Anchor Link

• An agent class of your own, with input and build.
• A graph of your own: a model node, a tool node, a request for approval, an effect node and two plain steps, connected by routers.
• A request that waits for a person, with a schema for the answer.
• An effect step that records its intent first, and a recovery that looks up the outcome.
• A failure in the middle of a payment, and a resume that pays once.

The complete code is in examples/tutorials/05-refund-graph.

Next, in tutorial 6, you serve the agent of tutorial 3 over HTTP.