Introduction
“AI agent” doesn’t have to mean a mysterious black box. In production, an agent is just a small loop that (1) reads a request, (2) proposes tool calls, (3) verifies preconditions and policy, (4) executes with receipts, and (5) returns a structured answer. Below is a complete, runnable example for a real workflow—an Order Return & Refund Support Agent—including a compact prompt contract, tool interfaces, and Python code that mediates actions safely.
The Use Case
A customer says an item arrived damaged and wants a refund. The agent should: check the order, validate return eligibility, create a return label, and—if policy allows—issue a refund. It must never claim success unless the system actually executed the action. It should cite the policy facts it used and show receipts for any state change.
The Prompt Contract (the agent’s “interface” to the model)
# file: contracts/support_refund_v1.yaml
role: "OrderCareAgent"
scope: >
Diagnose the user's issue about orders/returns. Use eligible claims for facts.
Ask once if critical fields are missing (order_id, item_id, reason, region).
Propose tool calls; never assert success unless a receipt is present.
output:
type: object
required: [summary, next_steps, citations, tool_proposals]
properties:
summary: {type: string, maxWords: 80}
next_steps: {type: array, items: {type: string, maxWords: 18}, maxItems: 6}
citations: {type: array, items: {type: string}} # claim_ids
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name: {type: string, enum: [GetOrder, CheckReturnPolicy, CreateReturnLabel, IssueRefund]}
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "returns_policy.us.v3"
citation_rule: "1-2 minimal-span claim_ids per factual sentence"
decoding:
narrative: {top_p: 0.92, temperature: 0.75, stop: ["\n\n## "]}
bullets: {top_p: 0.82, temperature: 0.45}
Example “claims” (what the model sees as context)
[
{"claim_id":"policy:return_window:30d","text":"Returns allowed within 30 days of delivery.",
"effective_date":"2025-01-10","source_id":"doc:return_policy_v3","span":"within 30 days"},
{"claim_id":"policy:refund:damage_evidence","text":"Refunds for damaged items require photo proof or carrier damage note.",
"effective_date":"2025-04-02","source_id":"doc:return_policy_v3","span":"photo proof or carrier note"},
{"claim_id":"order:1234:delivered_on","text":"Order 1234 delivered on 2025-10-05.",
"effective_date":"2025-10-05","source_id":"sys:orders","span":"delivered on 2025-10-05"}
]
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from datetime import datetime
from typing import Optional, List
class GetOrderArgs(BaseModel):
order_id: str
class CheckReturnPolicyArgs(BaseModel):
order_id: str
item_id: str
reason: str # e.g., "damaged"
delivered_on: datetime
class CreateReturnLabelArgs(BaseModel):
order_id: str
item_id: str
pickup: bool = False
class IssueRefundArgs(BaseModel):
order_id: str
item_id: str
amount_cents: int
reason: str
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str # stable receipt/id
message: str = ""
data: Optional[dict] = None
Each adapter must be idempotent: the same args + idempotency key return the same receipt without double-charging.
# adapters.py (mock business logic for demo purposes)
from tools import *
from datetime import datetime, timedelta
ORDERS = {
"1234": {"delivered_on": datetime(2025,10,5), "items": {"A-19": {"price_cents": 2599}}},
}
def get_order(a: GetOrderArgs) -> ToolReceipt:
if a.order_id in ORDERS:
return ToolReceipt(tool="GetOrder", ok=True, ref=f"ord-{a.order_id}",
data=ORDERS[a.order_id], message="Order found")
return ToolReceipt(tool="GetOrder", ok=False, ref=f"ord-{a.order_id}", message="Not found")
def check_return_policy(a: CheckReturnPolicyArgs) -> ToolReceipt:
if a.order_id not in ORDERS:
return ToolReceipt(tool="CheckReturnPolicy", ok=False, ref="pol-none", message="Order missing")
within_30 = (datetime.utcnow() - a.delivered_on) <= timedelta(days=30)
if not within_30:
return ToolReceipt(tool="CheckReturnPolicy", ok=False, ref="pol-expired", message="Outside 30-day window")
if a.reason.lower() != "damaged":
return ToolReceipt(tool="CheckReturnPolicy", ok=False, ref="pol-reason", message="Eligible: exchange only")
return ToolReceipt(tool="CheckReturnPolicy", ok=True, ref="pol-ok", message="Eligible for refund")
def create_return_label(a: CreateReturnLabelArgs) -> ToolReceipt:
return ToolReceipt(tool="CreateReturnLabel", ok=True, ref=f"rtn-{a.order_id}-{a.item_id}",
data={"label_url": f"https://label/{a.order_id}/{a.item_id}.pdf"})
def issue_refund(a: IssueRefundArgs) -> ToolReceipt:
return ToolReceipt(tool="IssueRefund", ok=True, ref=f"rfnd-{a.order_id}-{a.item_id}",
data={"amount_cents": a.amount_cents})
The Agent Loop (proposal → verification → execution → receipts)
# agent.py
import uuid, json
from typing import Dict, Any, List
from pydantic import BaseModel, ValidationError
from tools import *
from adapters import *
from datetime import datetime
# ---- Minimal policy gate ----
ALLOWED_TOOLS = {"GetOrder", "CheckReturnPolicy", "CreateReturnLabel", "IssueRefund"}
def verify_proposal(p: Dict[str, Any]) -> str:
"""Return '' if ok, otherwise reason string."""
required = {"name","args","preconditions","idempotency_key"}
if not required.issubset(set(p.keys())):
return "Missing fields in tool proposal"
if p["name"] not in ALLOWED_TOOLS:
return "Tool not allowed by policy"
# Example precondition guardrails
if p["name"] == "IssueRefund" and "amount_cents" in p["args"]:
if p["args"]["amount_cents"] <= 0:
return "Refund must be positive"
return ""
def execute(p: Dict[str, Any]) -> ToolReceipt:
name = p["name"]; args = p["args"]
if name == "GetOrder": return get_order(GetOrderArgs(**args))
if name == "CheckReturnPolicy": return check_return_policy(CheckReturnPolicyArgs(**args))
if name == "CreateReturnLabel": return create_return_label(CreateReturnLabelArgs(**args))
if name == "IssueRefund": return issue_refund(IssueRefundArgs(**args))
return ToolReceipt(tool=name, ok=False, ref="none", message="Unknown tool")
def new_idem() -> str:
return f"idem-{uuid.uuid4()}"
# ---- Model I/O shims (replace with your LLM client) ----
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], user_msg: str) -> Dict[str,Any]:
"""
For demo: produce a plausible tool plan.
Replace with your LLM call; ensure it returns JSON per contract.
"""
return {
"summary": "Item A-19 was damaged; eligible for refund within 30 days.",
"next_steps": ["Create a return label", "Issue refund once label is generated"],
"citations": ["policy:return_window:30d","policy:refund:damage_evidence","order:1234:delivered_on"],
"tool_proposals": [
{"name":"GetOrder","args":{"order_id":"1234"},"preconditions":"Order id provided.","idempotency_key": new_idem()},
{"name":"CheckReturnPolicy","args":{"order_id":"1234","item_id":"A-19","reason":"damaged",
"delivered_on": ORDERS["1234"]["delivered_on"].isoformat()},
"preconditions":"Within 30 days & reason=damaged.","idempotency_key": new_idem()},
{"name":"CreateReturnLabel","args":{"order_id":"1234","item_id":"A-19","pickup":False},
"preconditions":"Policy check ok.","idempotency_key": new_idem()},
{"name":"IssueRefund","args":{"order_id":"1234","item_id":"A-19","amount_cents":2599,
"reason":"damaged"},
"preconditions":"Return label created; policy ok.","idempotency_key": new_idem()}
]
}
def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
r_index = {r.tool:r for r in receipts}
lines = []
lines.append(model_json["summary"])
lines.append("")
lines.append("Next:")
for s in model_json["next_steps"]:
lines.append(f"• {s}")
lines.append("")
# reflect actual outcomes, not promises
if "CreateReturnLabel" in r_index and r_index["CreateReturnLabel"].ok:
lines.append(f"Return label: {r_index['CreateReturnLabel'].data['label_url']}")
if "IssueRefund" in r_index and r_index["IssueRefund"].ok:
amt = r_index["IssueRefund"].data["amount_cents"]/100
lines.append(f"Refund issued: ${amt:.2f} (ref {r_index['IssueRefund'].ref})")
lines.append("")
lines.append("Citations: " + ", ".join(model_json["citations"]))
return "\n".join(lines)
def handle(user_msg: str) -> str:
contract = open("contracts/support_refund_v1.yaml").read()
claims = json.loads(open("claims.json","w+").write("[]") or "[]") # plug real claims loader here
plan = call_model(contract, claims, user_msg)
receipts: List[ToolReceipt] = []
for prop in plan["tool_proposals"]:
reason = verify_proposal(prop)
if reason:
receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason))
continue
# Normalize datetime args if present
if "delivered_on" in prop["args"]:
prop["args"]["delivered_on"] = datetime.fromisoformat(prop["args"]["delivered_on"])
rec = execute(prop)
receipts.append(rec)
if not rec.ok and prop["name"] in {"CreateReturnLabel","IssueRefund"}:
# halt on critical failure; model would be re-queried with failure context in a full system
break
return render_response(plan, receipts)
if __name__ == "__main__":
print(handle("My order 1234 item A-19 arrived damaged. Can I return it and get a refund?"))
Run the script and you’ll get a response that (a) summarizes, (b) shows next steps, (c) includes a real return-label URL from the adapter, (d) includes a refund receipt only if the tool really ran, and (e) carries citations to the policy/order claims.
The Prompt You’d Send to the Model (compact and testable)
Below is a simplified system+user prompt that matches the contract above. In production you’d serialize the YAML contract, attach the claims, and require JSON output.
System:
You are OrderCareAgent. Follow the contract:
- Diagnose, then ask once if order_id, item_id, reason, or region is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, next_steps[], citations[], tool_proposals[].
Claims (eligible only):
[... JSON array of claim records like the examples above ...]
User:
My order 1234 item A-19 arrived damaged. Can I return it and get a refund?
Expected (model) JSON outline:
{
"summary": "Item A-19 was damaged; eligible for refund within 30 days.",
"next_steps": ["Create a return label", "Issue refund once label is generated"],
"citations": ["policy:return_window:30d","policy:refund:damage_evidence","order:1234:delivered_on"],
"tool_proposals": [
{"name":"GetOrder","args":{"order_id":"1234"},"preconditions":"Order id provided.","idempotency_key":"..."},
{"name":"CheckReturnPolicy","args":{"order_id":"1234","item_id":"A-19","reason":"damaged","delivered_on":"2025-10-05T00:00:00"},
"preconditions":"Within 30 days & reason=damaged.","idempotency_key":"..."},
{"name":"CreateReturnLabel","args":{"order_id":"1234","item_id":"A-19"},"preconditions":"Policy check ok.","idempotency_key":"..."},
{"name":"IssueRefund","args":{"order_id":"1234","item_id":"A-19","amount_cents":2599,"reason":"damaged"},
"preconditions":"Return label created; policy ok.","idempotency_key":"..."}
]
}
How to Extend This Pattern Quickly
Replace the mock adapters with your real order, label, and payment APIs; keep idempotency keys.
Load claims from your policy docs and order DB; enforce freshness windows before passing them in.
Add a validator step between call_model and execute: verify JSON schema, tone/lexicon, locale casing, citation coverage, and “no implied writes” language.
Move the YAML contract, policy bundle, and decoder settings into a “bundle” and ship behind a feature flag + canary, with one-click rollback.
Closing Thoughts
An AI agent is powerful when it’s boring: short prompts that define an interface, typed tool calls, explicit receipts, and evidence you can click. The example above is small by design so you can adapt it quickly to your stack—swap tools, swap policies, keep the safety rails. If you want a variant (Node, FastAPI, LangChain/LlamaIndex wiring, or adding human approvals), say the word and I’ll drop in a tailored version.
Join the conversation! Your thoughts help the community grow.