Example:
Enforcing tool-call order in LangGraph via in-tool validation.
Compliance rule: a user can only be enrolled in a plan AFTER
- the available plans have been listed, and
- the details of that specific plan have been reviewed.
The enforcement lives inside the enroll_in_plan tool, right next to the side effect, so it cannot be bypassed regardless of what order the model calls tools in. Prerequisites are tracked as plain state props that the upstream tools fill in as they run – tracked PER PLAN, not as a global flag.
from typing import Annotated
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition, InjectedState
from langgraph.types import Command
from langchain_core.tools import tool, InjectedToolCallId
from langchain_core.messages import ToolMessage, SystemMessage
# --- demo data -------------------------------------------------------------
_PLANS = [
{"id": "bronze", "name": "Bronze", "monthly": 120, "deductible": 6000,
"summary": "Low premium, high deductible. Good if you rarely need care."},
{"id": "silver", "name": "Silver", "monthly": 240, "deductible": 3000,
"summary": "Balanced premium and deductible."},
{"id": "gold", "name": "Gold", "monthly": 410, "deductible": 1000,
"summary": "High premium, low deductible. Good for frequent care."},
]
_PLANS_BY_ID = {p["id"]: p for p in _PLANS}
# --- state -----------------------------------------------------------------
def merge_plan_ids(left: set[str] | None, right: set[str] | None) -> set[str]:
"""Reducer: union, so concurrent/repeated detail views merge instead of
clobbering each other (and never raise INVALID_CONCURRENT_GRAPH_UPDATE)."""
return (left or set()) | (right or set())
class State(MessagesState):
# MessagesState already gives us `messages` with the add_messages reducer.
listed_plans: bool
reviewed_plan_ids: Annotated[set[str], merge_plan_ids]
enrolled_plan_id: str
# --- tools -----------------------------------------------------------------
@tool
def list_insurance_plans(
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
"""List all available insurance plans the user can choose from."""
summary = "\n".join(
f"- {p['id']}: {p['name']} (${p['monthly']}/mo, ${p['deductible']} deductible)"
for p in _PLANS
)
return Command(update={
"listed_plans": True,
"messages": [ToolMessage(f"Available plans:\n{summary}",
tool_call_id=tool_call_id)],
})
@tool
def get_plan_details(
plan_id: str,
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
"""Show the full details of a specific plan so the user can review it
before enrolling. `plan_id` must be one of the listed plan ids."""
plan = _PLANS_BY_ID.get(plan_id)
if plan is None:
return Command(update={"messages": [ToolMessage(
f"No plan with id '{plan_id}'. Call list_insurance_plans for valid ids.",
tool_call_id=tool_call_id)]})
details = (f"{plan['name']} ({plan['id']}): ${plan['monthly']}/mo, "
f"deductible ${plan['deductible']}.\n{plan['summary']}")
return Command(update={
# record WHICH plan was reviewed -- this is the compliance fact
"reviewed_plan_ids": {plan_id},
"messages": [ToolMessage(details, tool_call_id=tool_call_id)],
})
@tool
def enroll_in_plan(
plan_id: str,
state: Annotated[dict, InjectedState],
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
"""Enroll the user in a plan. Only valid after the plans have been listed
AND the details of this specific plan have been reviewed."""
# --- the guard: this is the real enforcement point ---
if not state.get("listed_plans"):
return _refuse(tool_call_id,
"Cannot enroll yet -- the available plans haven't been shown. "
"Call list_insurance_plans first.")
if plan_id not in state.get("reviewed_plan_ids", set()):
return _refuse(tool_call_id,
f"Cannot enroll in '{plan_id}' yet -- its details haven't been "
f"reviewed. Call get_plan_details(plan_id='{plan_id}') first, then enroll.")
if plan_id not in _PLANS_BY_ID:
return _refuse(tool_call_id, f"No plan with id '{plan_id}'.")
# prerequisites satisfied -> perform the privileged action
return Command(update={
"enrolled_plan_id": plan_id,
"messages": [ToolMessage(
f"Enrolled in the {_PLANS_BY_ID[plan_id]['name']} plan. "
"A confirmation email is on the way.",
tool_call_id=tool_call_id)],
})
def _refuse(tool_call_id: str, message: str) -> Command:
"""Fail helpfully: return a guiding ToolMessage instead of raising, so the
model reads what's missing and self-corrects by calling the right tool."""
return Command(update={"messages": [ToolMessage(message, tool_call_id=tool_call_id)]})
TOOLS = [list_insurance_plans, get_plan_details, enroll_in_plan]
SYSTEM_PROMPT = (
"You are an insurance enrollment assistant. Help the user choose and enroll "
"in a plan. You must show the available plans and the details of the chosen "
"plan before enrolling. If a tool tells you a prerequisite is missing, call "
"the tool it names, then retry."
)
# --- graph -----------------------------------------------------------------
def build_graph(model):
"""`model` is any chat model. Bind tools with parallel calls disabled so
`enroll_in_plan` can never be issued in the same step as its prerequisite
(tools in one superstep read a snapshot and wouldn't see each other's writes).
For OpenAI/Anthropic via LangChain: model.bind_tools(TOOLS, parallel_tool_calls=False)
"""
llm_with_tools = model.bind_tools(TOOLS)
def llm_call(state: State):
return {"messages": [llm_with_tools.invoke(
[SystemMessage(content=SYSTEM_PROMPT)] + state["messages"])]}
builder = StateGraph(State)
builder.add_node("llm_call", llm_call)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", tools_condition) # -> "tools" or END
builder.add_edge("tools", "llm_call")
return builder.compile()
if __name__ == "__main__":
# Requires an API key, e.g.:
# from langchain_anthropic import ChatAnthropic
# model = ChatAnthropic(model="claude-sonnet-4-6")
# graph = build_graph(model)
# for ev in graph.stream({"messages": [("user", "enroll me in the gold plan")]},
# stream_mode="values"):
# ev["messages"][-1].pretty_print()
print("Import this module and call build_graph(model). See test_guard.py for a "
"model-free check of the enforcement logic.")