Enforce an AI agent's next step in code

AI Agentscode-enforced AI conversation flow

Return the next allowed conversation state from code instead of letting the AI agent choose it.

governance

The claim

No slot is booked until a bike the shop actually services has been recorded. The rule lives in the tool handlers, which read collected state instead of asking the model whether it collected anything. The caller may well reach the scheduling step; what they cannot do is get a booking out of it.

Why it holds

valid_steps shapes the navigation tool the model is offered. It does not constrain your webhook, so the handler is the last authority on the transition. It is also not a lock, which is why the tool at the destination checks the same state again.

How it works

Two things move the conversation, and only one of them is clamped.

The model’s path is the next_step tool, bounded by valid_steps on the current step. The other path is a handler emitting a step change, and the SDK is explicit that this bypasses the clamp. Anything the handler emits happens. Neither path makes a step a boundary. A step is a place in a flow, so the tool that books the slot checks for a recorded bike whatever route reached it.

So the handler checks first:

def start_scheduling(self, args, raw_data):
    recorded = (raw_data or {}).get("global_data", {}).get("bike_type")
    if recorded not in SERVICEABLE:
        return FunctionResult(
            "NOT_READY: no serviceable bike has been recorded yet. Ask "
            "what kind of bike it is and call record_bike first."
        )
    return FunctionResult(...).swml_change_step("schedule")

The model may call start_scheduling whenever it likes, including on the first turn. What it cannot do is make the check pass. SERVICEABLE is a set in code, not a line in the prompt, so a caller who insists cannot widen it.

The refusal is prescriptive. It names the state that is missing and the tool that fills it, so the model’s next turn is a question rather than an apology.

One naming trap: swml_change_step() is the SDK method, and the key the platform receives is change_step. Assert the key.

Limitations

This governs one transition. A flow with several gates wants the check factored out, because a rule copied into four handlers is a rule that will disagree with itself.

Reading global_data from raw_data trusts the platform’s own state, not the model’s account of it. That is the point, but it means the value has to have been written by a handler.

What to change first

Delete the recorded not in SERVICEABLE check and keep the prompt instruction. The transition becomes advisory, which is the failure this recipe exists to prevent.

Where this sits

python/app.py

"""Enforce state transitions in a tool handler.

`valid_steps` shapes the navigation tool the model is offered. It does not
constrain your webhook, and it is not a lock: a step change from a handler
bypasses it. The handler is the last authority on whether the caller moves on,
and the tool that finally acts checks the state again rather than trusting
where it sits in the flow.

This agent takes a repair booking. The caller cannot reach scheduling until a
bike has actually been identified, and the check reads collected state rather
than asking the model whether it collected it.

Written against signalwire-sdk 3.0.1.
"""
import os

from dotenv import load_dotenv
from signalwire import AgentBase, FunctionResult

# the SDK does not read .env for you
load_dotenv()

# What the shop can actually service. Not in the prompt: the model cannot
# widen it by being persuaded.
SERVICEABLE = {"road", "gravel", "hybrid", "mountain"}


class BookingAgent(AgentBase):
    def __init__(self):
        super().__init__(name="booking", route="/booking")
        self.prompt_add_section(
            "Role",
            "You book repair appointments for a bicycle shop. Find out what "
            "kind of bike the caller has, then arrange a drop-off time.",
        )
        contexts = self.define_contexts()
        flow = contexts.add_context("default")

        identify = flow.add_step("identify_bike")
        identify.set_text(
            "Ask what kind of bike the caller needs repaired, and confirm it "
            "back to them before going further."
        )
        identify.set_functions(["record_bike", "start_scheduling"])
        # No native route out. Listing "schedule" here would hand the model a
        # next_step straight past the check below, which is the whole point.
        identify.set_valid_steps([])

        schedule = flow.add_step("schedule")
        schedule.set_text(
            "Offer a drop-off slot for the ${bike_type} the caller described."
        )
        schedule.set_functions(["confirm_slot"])
        schedule.set_valid_steps([])

    @AgentBase.tool(
        name="record_bike",
        description=(
            "Record the kind of bike the caller needs repaired. Call this as "
            "soon as they say what they ride."
        ),
        parameters={
            "type": "object",
            "properties": {
                "bike_type": {
                    "type": "string",
                    "description": (
                        "The kind of bike: road, gravel, hybrid or mountain."
                    ),
                }
            },
            "required": ["bike_type"],
        },
    )
    def record_bike(self, args, raw_data):
        kind = (args.get("bike_type") or "").strip().lower()
        if kind not in SERVICEABLE:
            # Refused, with the reason and what to do instead.
            return FunctionResult(
                f"UNSUPPORTED: the shop does not service {kind or 'that'}. "
                f"Tell the caller which types are serviced "
                f"({', '.join(sorted(SERVICEABLE))}) and ask again."
            )
        # State is written by code, from the validated value.
        return FunctionResult(
            f"Recorded a {kind}."
        ).update_global_data({"bike_type": kind})

    @AgentBase.tool(
        name="start_scheduling",
        description=(
            "Move on to arranging a drop-off time. Call this once the caller's "
            "bike has been recorded."
        ),
        parameters={"type": "object", "properties": {}},
    )
    def start_scheduling(self, args, raw_data):
        # The model may call this whenever it likes. Read the state instead
        # of believing it.
        recorded = (raw_data or {}).get("global_data", {}).get("bike_type")
        if recorded not in SERVICEABLE:
            return FunctionResult(
                "NOT_READY: no serviceable bike has been recorded yet. Ask "
                "what kind of bike it is and call record_bike first."
            )
        return FunctionResult(
            f"Scheduling a drop-off for the {recorded}."
        ).swml_change_step("schedule")

    @AgentBase.tool(
        name="confirm_slot",
        description="Confirm the drop-off slot the caller chose.",
        parameters={
            "type": "object",
            "properties": {
                "slot": {"type": "string",
                         "description": "The slot, such as 'Thursday 9am'."}
            },
            "required": ["slot"],
        },
    )
    def confirm_slot(self, args, raw_data):
        # Reaching this step is not permission to book. A step is a place in a
        # flow, not a security boundary, so the tool that acts checks the state
        # it depends on rather than trusting how the call arrived.
        recorded = (raw_data or {}).get("global_data", {}).get("bike_type")
        if recorded not in SERVICEABLE:
            return FunctionResult(
                "NOT_READY: no serviceable bike has been recorded, so there is "
                "nothing to book. Ask what kind of bike it is and call "
                "record_bike first."
            )
        return FunctionResult(f"Booked a {recorded} for {args.get('slot')}.")


agent = BookingAgent()

if __name__ == "__main__":
    agent.serve(host="0.0.0.0", port=int(os.getenv("PORT", "3000")))

Run it

cd python
pip install -r requirements.txt
cp ../.env.example .env          # set SWML_BASIC_AUTH_PASSWORD
python app.py

Point a phone number’s SWML webhook at https://<user>:<password>@<your-host>/booking/, using the credentials you set. Without them the request is refused.

Verify it

No network, no account:

python verify.py          # from the recipe folder, not python/

It renders the SWML and runs the handlers, asserting:

identify_bike exposes two tools and offers the model no next_step
an unserviceable bike is refused and writes no state
a serviceable one is normalised and written as set_global_data
start_scheduling is refused with no state, and refused again when the state is present but unserviceable
with real state it emits {"change_step": "schedule"}, and the SDK method name appears nowhere in the payload
confirm_slot refuses with no recorded bike, however the call reached it

Report an issue

Enforce an AI agent's next step in code

The claim

Why it holds

How it works

Limitations

What to change first

Where this sits

Seen in a build

Seen in a build

Seen in a build