Building an Agent Workflow

# Production-grade refund processing workflow
class RefundWorkflow:
    def __init__(self):
        self.nodes = {
            "lookup_order":    self._lookup_order,
            "check_eligibility": self._check_eligibility,
            "request_approval":  self._request_approval,
            "issue_refund":      self._issue_refund,
            "reject_refund":     self._reject_refund,
            "send_confirmation": self._send_confirmation,
        }
        self.edges = {
            "START":             {"*":        "lookup_order"},
            "lookup_order":      {"found":    "check_eligibility",
                                  "not_found":"reject_refund"},
            "check_eligibility": {"auto":     "issue_refund",
                                  "manual":   "request_approval",
                                  "ineligible":"reject_refund"},
            "request_approval":  {"approved": "issue_refund",
                                  "rejected": "reject_refund"},
            "issue_refund":      {"success":  "send_confirmation",
                                  "error":    "escalate"},
            "reject_refund":     {"*":        "send_confirmation"},
            "send_confirmation": {"*":        "END"},
        }

    def _lookup_order(self, state):
        oid = state["input"]["order_id"]
        db = {"ORD-123": {"total": 49.99, "days": 5, "email": "user@ex.com"}}
        order = db.get(oid)
        if order:
            state["data"]["order"] = order
            return "found"
        state["data"]["message"] = f"Order {oid} not found"
        return "not_found"

    def _check_eligibility(self, state):
        order = state["data"]["order"]
        if order["days"] > 30:
            state["data"]["message"] = "Outside 30-day return window"
            return "ineligible"
        return "auto" if order["total"] < 100 else "manual"

    def _request_approval(self, state):
        print(f"  >> Approval required for $" + str(state['data']['order']['total']))
        return "approved"  # Simulated approval

    def _issue_refund(self, state):
        state["data"]["refund_id"] = "REF-789"
        state["data"]["message"] = "Refund issued successfully"
        return "success"

    def _reject_refund(self, state):
        state["data"]["message"] = state["data"].get("message", "Refund rejected")
        return "*"

    def _send_confirmation(self, state):
        print(f"  Email: {state['data']['message']}")
        return "*"

    def run(self, order_id: str, reason: str) -> dict:
        state = {"input": {"order_id": order_id, "reason": reason}, "data": {}}
        current = self.edges["START"]["*"]
        steps = 0

        while current not in ("END", "escalate") and steps < 10:
            print(f"[{steps+1}] {current}")
            condition = self.nodes[current](state)
            current = self.edges.get(current, {}).get(condition,
                       self.edges.get(current, {}).get("*", "END"))
            steps += 1

        return {"status": "complete", "steps": steps, "result": state["data"].get("message")}

wf = RefundWorkflow()
result = wf.run("ORD-123", "Product damaged")
print(f"\nResult: {result}")