Plan-and-Execute Agents in Python: Beyond the ReAct Loop

You describe each step as a node in a graph, wire the nodes together, and let an engine walk the graph. In Promptise Foundry the graph is a PromptGraph, the steps are nodes such as PlanNode, PromptNode and ValidateNode, and PromptGraphEngine runs it:

        tools = agent.tools
        graph = PromptGraph("plan-and-execute", mode="static")
        graph.add_node(PlanNode("plan", output_schema=Plan, instructions=f"Plan the steps. Tools: {[t.name for t in tools]}"))
        graph.add_node(PromptNode("act", tools=tools))
        graph.add_node(ValidateNode("verify", output_schema=Verdict, on_pass="__end__", on_fail="act"))
        graph.sequential("plan", "act", "verify")
        graph.set_entry("plan")

        engine = PromptGraphEngine(graph=graph, model=resolve_model("openai:gpt-5-mini"), max_iterations=12)
        result = await engine.ainvoke({"messages": [{"role": "user", "content": "Which regions missed their Q3 target?"}]})

The planner writes subgoals, the act node works through them with your MCP tools, and the verifier either ends the run or sends the work back. If you only want a self-check and no new steps, there's a one-word version: build_agent(..., agent_pattern="verify"). The rest of this guide shows both running, adds a reflect step, and measures whether any of it was worth it.

Every Promptise agent is a graph, even the default one. build_agent gives you a ReAct graph: one PromptNode named reason that holds the tools and loops until the model stops calling them. A custom pattern is just a bigger graph. Here's the one you'll build:

After each node, the engine picks the next one in a fixed order. If the node called tools, it runs the same node again so the model sees the results. Otherwise it takes, in this order: a next node the node chose itself (this is how ValidateNode routes on pass or fail), a route field in the node's structured output, your edges (always, sequential, when, on_output and friends), the node's transitions, and finally its default_next. If none applies, the run ends.

Promptise ships ten reasoning nodes that come with their own instructions: PlanNode, ThinkNode, ReflectNode, ObserveNode, JustifyNode, CritiqueNode, SynthesizeNode, ValidateNode, RetryNode and FanOutNode. Next to them sit the general ones, such as PromptNode, ToolNode, RouterNode, GuardNode, ParallelNode, HumanNode and the @node decorator, which turns any async function into a step. Nodes also take typed flags from NodeFlag, for example CRITICAL to abort the run if the node fails, RETRYABLE to retry it with backoff, or SKIP_ON_ERROR to skip it after a failed step.

You don't have to build a graph to change the pattern. build_agent takes a name:

• Python 3.10 or newer.

• Promptise Foundry from PyPI. Everything here uses the base install:

pip install promptise
export OPENAI_API_KEY="sk-..."

• An API key for a model provider. The examples use openai:gpt-5-mini; any model with tool calling and structured output works.

You need a task where a shortcut gives a wrong answer, or a check has nothing to catch. This MCP server has two tools: one returns each region's quarterly target, the other returns monthly gross revenue and refunds. The targets are for net revenue, so an agent that forgets the refunds will think two regions hit their target when they didn't.

from promptise.mcp.server import MCPServer

server = MCPServer("sales")

# A stand-in for your warehouse. Amounts are in USD.
TARGETS = {"Q3": {"North": 300_000, "South": 250_000, "East": 180_000, "West": 220_000}}

REVENUE = {
    ("North", "Q3"): [("Jul", 104_200, 3_100), ("Aug", 98_750, 2_400), ("Sep", 101_300, 4_050)],
    ("South", "Q3"): [("Jul", 86_400, 1_200), ("Aug", 82_100, 950), ("Sep", 88_900, 1_650)],
    ("East", "Q3"): [("Jul", 55_300, 2_900), ("Aug", 61_800, 3_300), ("Sep", 58_050, 2_150)],
    ("West", "Q3"): [("Jul", 74_900, 1_100), ("Aug", 71_250, 2_600), ("Sep", 76_400, 1_900)],
}


@server.tool()
async def get_targets(quarter: str) -> dict:
    """Get each region's net revenue target for a quarter.

    Args:
        quarter: The quarter, for example "Q3".
    """
    return {"quarter": quarter, "net_revenue_targets": TARGETS.get(quarter.upper(), {})}


@server.tool()
async def get_monthly_revenue(region: str, quarter: str) -> dict:
    """Get a region's gross revenue and refunds for each month of a quarter.

    Args:
        region: North, South, East or West.
        quarter: The quarter, for example "Q3".
    """
    rows = REVENUE.get((region.strip().title(), quarter.upper()))
    if rows is None:
        return {"error": f"No data for {region} in {quarter}."}
    return {
        "region": region.strip().title(),
        "quarter": quarter.upper(),
        "months": [{"month": m, "gross": g, "refunds": r} for m, g, r in rows],
    }


if __name__ == "__main__":
    server.run()

Computed straight from the data, this is the answer every agent has to reach. North and West beat their target on gross revenue and miss it on net:

North gross=304,250 refunds=9,550 net=294,700 target=300,000 gap=-5,300 (-1.77%)
South gross=257,400 refunds=3,800 net=253,600 target=250,000 gap=+3,600 (+1.44%)
East  gross=175,150 refunds=8,350 net=166,800 target=180,000 gap=-13,200 (-7.33%)
West  gross=222,550 refunds=5,600 net=216,950 target=220,000 gap=-3,050 (-1.39%)

Start with the baseline, so you know what any extra structure has to beat. To count model calls and tokens, pass a small LangChain callback in the run's config. It sees every model call, whichever pattern runs:

from langchain_core.callbacks import BaseCallbackHandler


class Usage(BaseCallbackHandler):
    """Count model calls and tokens for one run."""

    def __init__(self):
        self.calls = self.input_tokens = self.output_tokens = 0

    def on_llm_end(self, response, **kwargs):
        self.calls += 1
        usage = response.generations[0][0].message.usage_metadata or {}
        self.input_tokens += usage.get("input_tokens", 0)
        self.output_tokens += usage.get("output_tokens", 0)

    def __str__(self):
        return f"{self.calls} model calls, {self.input_tokens:,} input + {self.output_tokens:,} output tokens"

import asyncio
import sys

from promptise import build_agent
from promptise.config import StdioServerSpec

from usage import Usage

QUESTION = "Which regions missed their Q3 revenue target, and by how much in dollars and percent?"


async def main():
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={"sales": StdioServerSpec(command=sys.executable, args=["sales_server.py"])},
        instructions="You are a sales analyst. Use the sales tools for every number.",
        agent_pattern="react",  # the default
        trace_tools=True,
    )
    try:
        usage = Usage()
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": QUESTION}]},
            config={"callbacks": [usage]},
        )
        print(">>>", result["messages"][-1].content)
        print("path:", " -> ".join(agent.last_report.nodes_visited))
        print("usage:", usage)
    finally:
        await agent.shutdown()


asyncio.run(main())

→ Invoking tool: get_targets with {'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'North', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'South', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'East', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'West', 'quarter': 'Q3'}
…
>>> Summary — regions that missed Q3 targets

Method: net Q3 revenue = sum(gross − refunds) for Jul–Sep; compared to the Q3 net revenue targets.

- North: Net = $294,700 vs target $300,000 → missed by $5,300 (1.8% below target).
- East: Net = $166,800 vs target $180,000 → missed by $13,200 (7.3% below target).
- West: Net = $216,950 vs target $220,000 → missed by $3,050 (1.4% below target).

Note: South hit/beat its target — Net = $253,600 vs target $250,000 (exceeded by $3,600, 1.4% above target).
path: reason -> reason
usage: 2 model calls, 1,063 input + 1,133 output tokens

Two model calls. In the first, the model asked for all five tool calls at once, and Promptise ran them in parallel. In the second, it wrote a correct answer, refunds and all. agent.last_report is the engine's execution report; nodes_visited is the path through the graph; for this graph, each entry is one model call.

Before you build anything, try the one-line change. With agent_pattern="verify", the same single node is told to plan, solve, and check its own work in one response:

        agent_pattern="verify",

>>> PLAN
- Restate: Determine which regions failed to meet their Q3 net revenue targets and quantify each shortfall in dollars and percent.
…
VERIFY
- Rechecked by summing monthly net (gross − refunds) per month:
  - North per-month nets: 101,100 + 96,350 + 97,250 = 294,700 (matches)
…
FINAL ANSWER
Regions that missed their Q3 targets:
- North: missed by $5,300 (1.77% below target).
- East: missed by $13,200 (7.33% below target).
- West: missed by $3,050 (1.39% below target).
…
path: reason -> reason
usage: 2 model calls, 1,201 input + 2,044 output tokens

Same two calls, correct again, and nearly twice the output tokens. Notice where the self-check lives: in the answer. Your users see the plan and the arithmetic, so this pattern suits an analyst's report better than a chat reply.

Now the real pattern. Four nodes, each with one job:

from typing import TypedDict

from promptise.engine import PlanNode, PromptGraph, PromptNode, ReflectNode, ValidateNode


class Plan(TypedDict):
    subgoals: list[str]
    quality_score: int


class Verdict(TypedDict):
    passes: bool
    issues: list[str]


def plan_act_verify(tools, system_prompt: str = "") -> PromptGraph:
    """Plan, act with tools, verify the draft, and reflect before retrying."""
    tool_list = "\n".join(f"- {t.name}: {t.description.splitlines()[0]}" for t in tools)
    graph = PromptGraph("plan-act-verify", mode="static")

    graph.add_node(
        PlanNode(
            "plan",
            output_schema=Plan,
            instructions=(
                f"{system_prompt}\n\nPlan how to answer the question with these tools:\n{tool_list}\n\n"
                "Write 4 or fewer subgoals, each one step. Never plan to ask the user for data "
                "a tool can fetch. Rate the plan from 1 to 5 as quality_score."
            ),
        )
    )
    graph.add_node(
        PromptNode(
            "act",
            instructions=f"{system_prompt}\n\nWork through the plan with your tools, then write the answer.",
            tools=tools,
        )
    )
    graph.add_node(
        ValidateNode(
            "verify",
            criteria=[
                "The answer names every region that missed, with the shortfall in dollars and percent",
                "Every number in the answer matches the tool results",
                "Every sum, difference and percentage is recomputed and correct",
                "The answer uses the same definition as the targets (net, not gross)",
            ],
            output_schema=Verdict,
            on_pass="__end__",
            on_fail="reflect",
        )
    )
    graph.add_node(ReflectNode("reflect", input_keys=["validation"]))

    graph.sequential("plan", "act", "verify")
    graph.always("reflect", "act")
    graph.set_entry("plan")
    return graph

What each node does:

• `plan` is a PlanNode. It asks the model for subgoals and a quality score. When the output has subgoals, the node stores them as the run's plan, and a score below its quality_threshold (3 by default) sends it round to plan again. Later nodes see the plan in their prompt.

• `act` is an ordinary PromptNode with the MCP tools. It loops through tool calls exactly like the ReAct node, then writes a draft.

• `verify` is a ValidateNode. It checks the draft against your criteria and returns passes and issues. Its on_pass and on_fail decide where the run goes.

• `reflect` is a ReflectNode. It reads the verdict through input_keys=["validation"], validation being where ValidateNode stores its output, and writes down what went wrong before act tries again.

The edges are two lines: sequential chains plan, act and verify, and always sends reflect back to act.

build_agent connects to your MCP servers and discovers their tools. Hand those tools to the graph, and run it with PromptGraphEngine. A small hook prints each node as it finishes, so you can watch the path:

import asyncio
import sys

from promptise import build_agent, resolve_model
from promptise.config import StdioServerSpec
from promptise.engine import PromptGraphEngine

from plan_act_verify import plan_act_verify
from usage import Usage

QUESTION = "Which regions missed their Q3 revenue target, and by how much in dollars and percent?"
INSTRUCTIONS = "You are a sales analyst. Use the sales tools for every number."


class ShowSteps:
    """Engine hook: print each node as it finishes, with its structured output."""

    async def post_node(self, node, result, state):
        detail = result.output if isinstance(result.output, dict) else ""
        print(f"[{node.name}] {detail}")
        return result


async def main():
    # build_agent connects to the MCP servers and discovers their tools.
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={"sales": StdioServerSpec(command=sys.executable, args=["sales_server.py"])},
        trace_tools=True,
    )
    try:
        engine = PromptGraphEngine(
            graph=plan_act_verify(agent.tools, INSTRUCTIONS),
            model=resolve_model("openai:gpt-5-mini"),
            hooks=[ShowSteps()],
            max_iterations=12,  # a hard cap on node runs, retries included
        )
        usage = Usage()
        result = await engine.ainvoke(
            {"messages": [{"role": "user", "content": QUESTION}]},
            config={"callbacks": [usage]},
        )
        print(">>>", result["messages"][-1].content)
        print("path:", " -> ".join(engine.last_report.nodes_visited))
        print("usage:", usage)
    finally:
        await agent.shutdown()


asyncio.run(main())

[plan] {'subgoals': ['Fetch Q3 net revenue targets for all regions using get_targets.', 'For each region, fetch gross revenue and refunds for each month of Q3 using get_monthly_revenue.', "Compute each region's Q3 net revenue (gross - refunds summed), compare to target, and calculate dollar and percent shortfall; list regions that missed their target with amounts."], 'quality_score': 5}
→ Invoking tool: get_targets with {'quarter': 'Q3'}
✔ Tool result from get_targets: {"quarter": "Q3", "net_revenue_targets": {"North": 300000, "South": 250000, "East": 180000, "West": 220000}}
[act] 
→ Invoking tool: get_monthly_revenue with {'region': 'North', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'South', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'East', 'quarter': 'Q3'}
→ Invoking tool: get_monthly_revenue with {'region': 'West', 'quarter': 'Q3'}
…
[act] 
[act] 
[verify] {'passes': True, 'issues': []}
>>> Using Q3 targets and monthly revenue data:

- North — Target: $300,000; Q3 net revenue: $294,700 (Gross minus refunds). Missed by $5,300 (1.77% short).
- East — Target: $180,000; Q3 net revenue: $166,800. Missed by $13,200 (7.33% short).
- West — Target: $220,000; Q3 net revenue: $216,950. Missed by $3,050 (1.39% short).

(For context: South exceeded its target — Target $250,000; Q3 net revenue $253,600 — $3,600 above target, +1.44%.)
path: plan -> act -> act -> act -> verify
usage: 5 model calls, 3,309 input + 2,724 output tokens

Read the path: one call to plan, three rounds of act (targets first, then the four regions, then the answer), one call to verify. The plan made the agent more sequential: it fetched the targets in one round and the revenue in the next, where the ReAct agent did both at once. The answer is right, and the verifier passed it on the first try, so reflect never ran.

One run proves little, because models vary. So the same question went to each pattern ten times, with the answer checked automatically for all three dollar shortfalls. act-check is a fourth pattern you'll meet later in this guide. The model call and token counts come from the Usage callback above:

pattern           correct  model calls   tokens  median s
react               10/10          2.0    2,336      11.6
verify              10/10          2.0    3,664      17.5
plan-act-verify     10/10          4.9    6,281      22.4
act-check           10/10          2.0    2,375      10.4

Every pattern got every run right. That's the honest headline: with a capable model and a task this size, the ReAct loop doesn't make the mistake the extra steps are there to catch. The plan, act, verify graph paid for its structure anyway, with about 2.5 times the model calls, 2.7 times the tokens and twice the wait. Its verifier passed all ten drafts on the first try, so the reflect step never ran once. verify cost about half as much again as ReAct, most of it in a longer answer.

That doesn't make custom patterns useless. It means the value of a step is the mistakes it catches, and you only learn that by measuring on your own task.

The first version of this graph used PlanNode with its built-in instructions, which don't mention your tools. Most runs were fine. Then one planner, unaware that the agent had a sales database, planned to ask the user for the data:

[plan] {'subgoals': ['(Highest priority — First) Ask you to provide the Q3 revenue data: a table (CSV, Excel, or pasted) with columns exactly named: Region, Q3 Actual Revenue, Q3 Revenue Target.', …
[act] 
[verify] {'passes': False, 'issues': ["Did not confirm that the Q3 revenue targets and actuals are 'net' (not gross). …
[reflect] 
[act] 
[verify] {'passes': False, 'issues': ['No computations or results were produced: the assistant only requested data and described how it would compute shortfalls, …
…
[verify] {'passes': True, 'issues': ['No numeric outputs were produced, so the arithmetic- and number-matching criteria are vacuously satisfied …
>>> I can do that — I just need your Q3 data first.
…
path: plan -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify -> reflect -> act -> verify
usage: 24 model calls, 105,866 input + 29,815 output tokens

The act node followed the plan and never called a tool. The verifier rejected it seven times, each round came back with a more polished request for the data, and on the eighth pass the verifier gave up and passed an answer with no numbers in it, because no numbers meant nothing to get wrong. Twenty-four model calls and about 136,000 tokens bought a request for data the agent already had.

Three lessons came out of that run, and they're all in the final graph:

• Tell the planner what the tools are. A plan written blind is a guess. The fixed PlanNode gets the tool list and one rule: never plan to ask for data a tool can fetch.

• Make "there is an answer" a criterion. An LLM verifier judges what's in front of it. The first criterion now demands every missed region with both numbers.

• Cap the loop in the engine. That run passed max_iterations=3 to the verify node and it ran eight times: in 1.2.1 the engine doesn't enforce a node's own max_iterations. The limits that hold are the engine's max_iterations (all node runs, 50 by default) and max_node_iterations (runs of any one node, 25 by default).

A model checking a model is the expensive way to verify. When the answer can be computed, check it with code: it's free, it's fast, and it can't be talked into passing. The @node decorator turns an async function into a graph step. This one recomputes every shortfall from the raw tool results the engine keeps in state.observations, and sends the agent back with the right numbers if the draft is missing any:

import json

from langchain_core.messages import HumanMessage

from promptise.engine import NodeResult, PromptGraph, PromptNode, node


@node("check")
async def check(state):
    """Recompute every shortfall from the raw tool results and compare it with the draft."""
    targets, net = {}, {}
    for obs in state.observations:
        data = json.loads(obs["result"])
        if obs["tool"] == "get_targets":
            targets = data["net_revenue_targets"]
        elif obs["tool"] == "get_monthly_revenue":
            net[data["region"]] = sum(m["gross"] - m["refunds"] for m in data["months"])

    draft = state.messages[-1].content.replace(",", "")
    wrong = [
        f"{region} missed by ${targets[region] - value:,}"
        for region, value in net.items()
        if value < targets[region] and str(targets[region] - value) not in draft
    ]
    if not wrong:
        return NodeResult(node_name="check", next_node="__end__")

    state.messages.append(HumanMessage(f"Check failed. The answer must state: {'; '.join(wrong)}. Fix it."))
    return NodeResult(node_name="check", next_node="act", output={"wrong": wrong})


def act_then_check(tools, system_prompt: str = "") -> PromptGraph:
    """A ReAct node whose answer is checked by code before it leaves the graph."""
    graph = PromptGraph("act-check", mode="static")
    graph.add_node(PromptNode("act", instructions=system_prompt, tools=tools, default_next="check"))
    graph.add_node(check)
    graph.set_entry("act")
    return graph

Run it like the graph in Step 5, and the path is the ReAct path plus one step that costs nothing:

path: act -> act -> check
usage: 2 model calls, 1,063 input + 1,173 output tokens

The model got it right in every run, so to see the failure branch, feed the check a draft that made the classic mistake, using gross revenue instead of net:

next node: act
message added: Check failed. The answer must state: North missed by $5,300; East missed by $13,200; West missed by $3,050. Fix it.

The agent goes back to act with the exact numbers it has to explain. A check like this is specific to your task, and that's the point: you know what a correct answer looks like better than a second model does.

The Reasoning Graph is flexible, and in 1.2.1 a few corners are rough. Plan around these:

• `inject_tools=True` gets no tools from `build_agent`. Pass tools= to the node yourself and run the graph with PromptGraphEngine, as in Step 5.

• A node keeps its first system prompt for the whole run. When the engine enters a node again, it reuses the prompt it built the first time, so updated input_keys, plan progress and reflections don't reach it. In a test, a node fed "apple" and then "banana" answered "apple" twice. Pass feedback as a message instead, as ReflectNode and the code check above do.

• Routing on output needs a `dict`. Nodes that answer in plain text produce no output to route on, and pydantic schemas don't route either. Use a TypedDict.

• `agent_pattern="peoatr"` doesn't end on its own. Its nodes route on output fields their plain-text answers never produce, so on this task it cycled act, think and reflect until the 25-step cap: 26 model calls, about 88,600 tokens, and a final message that was the think node's notes, not an answer.

• `agent_pattern="pipeline"` raises TypeError: build_pipeline_graph() got an unexpected keyword argument 'tools'. Build a pipeline with PromptGraph.pipeline(node_a, node_b) instead.

• An unknown pattern name falls back to ReAct without a warning. A typo like agent_pattern="plan-act-verify" gives you the default graph.

• The execution report counts zero tokens. last_report.total_tokens stayed at 0 in our runs, because the engine reads the model's usage as attributes and LangChain returns it as a dict. Count with a callback like Usage above.

• Saving to YAML is a snapshot, not a round trip. save_graph from promptise.engine.serialization writes nodes, instructions and plain edges, but drops output_schema, on_pass and on_fail, and stores tools as names. Load it back and the verify step no longer routes.

Is plan and execute better than ReAct?

Not by default. On this task, ten runs each, ReAct was right every time with the fewest calls, and the plan, act, verify graph was right every time for about 2.7 times the tokens. Planning helps when the order of steps matters and the model gets it wrong without guidance, and it hurts when the planner can't see what the agent can do. Measure on your own task before you switch.

What is a ReAct agent?

An agent that alternates reasoning and acting: the model decides on a tool call, sees the result, and repeats until it can answer. The pattern comes from the 2022 paper ReAct: Synergizing Reasoning and Acting in Language Models. It's what build_agent gives you unless you choose otherwise.

What is a reflection agent?

An agent that reviews its own work and tries again with what it learned. Here, ReflectNode reads the verifier's issues and writes down the mistake and a correction before the act step runs again. Reflection only helps if something real triggers it, so pair it with a verifier that can actually fail, ideally one written in code.

Is the Promptise Reasoning Graph a LangGraph alternative?

It covers the same ground: an agent as a graph of steps with conditional edges and loops. The difference is what a node is. A Promptise node is a whole reasoning step with its own instructions, tools, output schema and routing, and ten of them come ready-made. Promptise builds on LangChain's model and tool interfaces, so LangChain tools and callbacks, like the Usage counter here, work unchanged.

Can I use a different model for some nodes?

Yes. Every PromptNode accepts model_override, and so do the reasoning nodes built on it, such as PlanNode and ValidateNode. It takes either a model object or a string such as "openai:gpt-5-mini". A common split is a small, fast model for planning and checking and a stronger one for the act step.

• Reasoning Graph: the engine, its two modes, and every building block.

• Nodes and Node Flags: parameters for every node type and what each flag does.

• Prebuilt Patterns and Reasoning Patterns: the built-in graphs and when to pick one.

• Building Custom Reasoning: more graph shapes, from pipelines to parallel research.

• Edges & Transitions and Hooks: conditional routing, budgets and per-node metrics.

• How to Connect MCP Servers to Your AI Agent in Python: where the tools in this guide come from.