Building an Approval & Refund Graph Workflow with ADK 2.0

This hands-on tutorial walks through constructing a complete graph-based refund processing workflow using Google ADK 2.0. By combining risk assessment logic with deterministic conditional routing, we ensure low-value, low-risk requests are auto-approved while suspicious or high-value claims are automatically gated for human manager review.

Workflow Architecture

The state graph follows a clean routing topology based on financial thresholds and evaluated fraud scores:

flowchart TD
    A["Risk Assessment Node"]

    B{"risk < 0.5<br/>AND<br/>amount ≤ $100"}

    C["Auto-Approve Node"]
    D["Human Review Gate Node"]

    A --> B
    B -- "Yes" --> C
    B -- "No" --> D

    classDef assessment fill:#e0f2fe,stroke:#0284c7,stroke-width:2px,color:#0f172a;
    classDef decision fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a;
    classDef success fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a;
    classDef review fill:#fee2e2,stroke:#dc2626,stroke-width:2px,color:#0f172a;

    class A assessment;
    class B decision;
    class C success;
    class D review;

Python Implementation

Here is the complete Python implementation using asynchronous workflow nodes and strongly-typed state objects:

import asyncio
from typing import Dict, Any

# 1. State Definition
class RefundWorkflowState:
    """Carries customer request data and processing status through the graph nodes."""
    def __init__(self, request_id: str, amount: float, reason: str):
        self.request_id = request_id
        self.amount = amount
        self.reason = reason
        self.risk_score: float = 0.0
        self.status: str = "PENDING"
        self.approver: str = "NONE"

# 2. Graph Node Functions
async def risk_assessment_node(state: RefundWorkflowState) -> RefundWorkflowState:
    """Node 1: Calculates fraud risk score based on amount and reason keywords."""
    print(f"[NODE: Risk Assessment] Evaluating request {state.request_id} for ${state.amount}")

    # Deterministic risk calculation rule
    if state.amount > 500.0 or "urgent" in state.reason.lower():
        state.risk_score = 0.85
    else:
        state.risk_score = 0.15

    return state

async def auto_approve_node(state: RefundWorkflowState) -> RefundWorkflowState:
    """Node 2A: Automatically approves low-risk, low-value refunds."""
    print(f"[NODE: Auto Approve] Auto-approving request {state.request_id}")
    state.status = "APPROVED"
    state.approver = "SYSTEM_AUTO_RULES"
    return state

async def human_review_gate_node(state: RefundWorkflowState) -> RefundWorkflowState:
    """Node 2B: Escalates high-risk or high-value refunds to manager review queue."""
    print(f"[NODE: Human Gate] Escalating request {state.request_id} to Manager Queue")
    state.status = "REQUIRES_HUMAN_APPROVAL"
    state.approver = "PENDING_MANAGER_SIGN_OFF"
    return state

# 3. Conditional Edge Routing Function
def route_after_risk_check(state: RefundWorkflowState) -> str:
    """Conditional Edge: Programmatically determines next graph node based on state."""
    if state.risk_score < 0.50 and state.amount <= 100.0:
        return "auto_approve_node"
    return "human_review_gate_node"

# 4. Graph Execution Simulation
async def run_refund_graph(state: RefundWorkflowState):
    print(f"\n--- Starting Refund Graph for Request: {state.request_id} ---")

    # Execute Node 1
    state = await risk_assessment_node(state)

    # Evaluate Conditional Edge
    next_node = route_after_risk_check(state)
    print(f"[ROUTER EDGE] Routing to: {next_node}")

    # Transition to Next Node
    if next_node == "auto_approve_node":
        state = await auto_approve_node(state)
    elif next_node == "human_review_gate_node":
        state = await human_review_gate_node(state)

    print(f"[WORKFLOW COMPLETE] Final Status: {state.status} | Approver: {state.approver}")

async def main():
    # Test Case 1: Low amount ($45.00), low risk -> Auto Approved
    req1 = RefundWorkflowState(request_id="REQ-001", amount=45.0, reason="Item defective")
    await run_refund_graph(req1)

    # Test Case 2: High amount ($750.00) -> Escalated to Human Review Gate
    req2 = RefundWorkflowState(request_id="REQ-002", amount=750.0, reason="Urgent refund needed")
    await run_refund_graph(req2)

if __name__ == "__main__":
    asyncio.run(main())

Running the Execution Simulation

Executing the script produces the following deterministic trace output:

--- Starting Refund Graph for Request: REQ-001 ---
[NODE: Risk Assessment] Evaluating request REQ-001 for $45.0
[ROUTER EDGE] Routing to: auto_approve_node
[NODE: Auto Approve] Auto-approving request REQ-001
[WORKFLOW COMPLETE] Final Status: APPROVED | Approver: SYSTEM_AUTO_RULES

--- Starting Refund Graph for Request: REQ-002 ---
[NODE: Risk Assessment] Evaluating request REQ-002 for $750.0
[ROUTER EDGE] Routing to: human_review_gate_node
[NODE: Human Gate] Escalating request REQ-002 to Manager Queue
[WORKFLOW COMPLETE] Final Status: REQUIRES_HUMAN_APPROVAL | Approver: PENDING_MANAGER_SIGN_OFF

Core Architectural Principles

[!TIP] Always decouple decision thresholds from prompt engineering when financial or security liabilities exist.

1. Deterministic Risk Assessment: The scoring rules rely on strict financial limits rather than stochastic LLM prompts. This makes the system predictable and legally auditable.
2. Pure Python Edge Functions: The route_after_risk_check function relies solely on state inspection. It can be unit-tested in isolation without mocking external APIs or running the LLM.
3. Isolated Async Nodes: Every node function is self-contained. You can swap implementations, inject telemetry, or connect database backends without modifying neighboring nodes.

Extending the Workflow Graph

In production, you can expand this basic graph topology by adding downstream operational nodes:

• notify_customer_node: Generates and dispatches status notification emails or SMS messages.
• audit_log_node: Writes immutable state transition logs to BigQuery or PostgreSQL.
• fraud_investigation_node: Automatically collects background user telemetry for flagged requests.

Each new node requires only a Python handler and a corresponding edge routing rule to integrate into the workflow chain.

Key Takeaway

Graph workflows empower developers to build complex agentic applications that combine the creative context understanding of LLMs with the absolute compliance safety of Python conditional edges.