Flows cheatsheet

from kapso.builder.flows import Flow
from kapso.builder.flows.nodes import *
from kapso.builder.flows.nodes.decide import Condition
from kapso.builder.ai.field import AIField

flow = (Flow(name="Customer Service")
    .add_node(StartNode(id="start"))
    .add_node(SendTextNode(
        id="greeting",
        message="Hi! How can I help you today?"
    ))
    .add_node(WaitForResponseNode(id="get_request"))
    .add_node(DecideNode(
        id="route",
        conditions=[
            Condition(label="support", description="Technical support needed"),
            Condition(label="sales", description="Sales or product question"),
            Condition(label="other", description="General inquiry")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="support_agent",
        system_prompt="You're a technical support agent. Help solve technical issues.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="sales_agent",
        system_prompt="You're a sales agent. Help with product questions and sales.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(SendTextNode(
        id="fallback",
        message="Thanks for your message. Someone will get back to you soon."
    ))

    .add_edge("start", "greeting")
    .add_edge("greeting", "get_request")
    .add_edge("get_request", "route", "response")
    .add_edge("route", "support_agent", "support")
    .add_edge("route", "sales_agent", "sales")
    .add_edge("route", "fallback", "other")
)

flow = (Flow(name="Check User First")
    .add_node(StartNode(id="start"))
    .add_node(FunctionNode(
        id="check_user",
        function_id="verify_user_phone",  # From: kapso functions list
        save_response_to="user"
    ))
    .add_node(DecideNode(
        id="route_user",
        conditions=[
            Condition(label="registered", description="User found in database"),
            Condition(label="new", description="New user, not registered")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(SendTextNode(
        id="welcome_back",
        message="Welcome back {{user.name}}! How can I help?"
    ))
    .add_node(SendTextNode(
        id="welcome_new",
        message="Welcome! Let's get you set up."
    ))

    .add_edge("start", "check_user")
    .add_edge("check_user", "route_user")
    .add_edge("route_user", "welcome_back", "registered")
    .add_edge("route_user", "welcome_new", "new")
)

flow = (Flow(name="Personalized Flow")
    .add_node(StartNode(id="start"))
    .add_node(FunctionNode(
        id="load_profile",
        function_id="get_user_profile",  # From: kapso functions list
        save_response_to="profile"
    ))
    .add_node(SendTextNode(
        id="personalized_greeting",
        message=AIField(prompt="Greet {{profile.name}} mentioning their {{profile.subscription_type}} plan"),
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(WaitForResponseNode(id="wait"))
    .add_node(AgentNode(
        id="personal_agent",
        system_prompt="Help {{profile.name}}. Plan: {{profile.subscription_type}}. Last order: {{profile.last_order}}",
        provider_model_name="claude-sonnet-4-20250514"
    ))

    .add_edge("start", "load_profile")
    .add_edge("load_profile", "personalized_greeting")
    .add_edge("personalized_greeting", "wait")
    .add_edge("wait", "personal_agent", "response")
)

flow = (Flow(name="Order Lookup")
    .add_node(StartNode(id="start"))
    .add_node(SendTextNode(id="ask_order", message="What's your order number?"))
    .add_node(WaitForResponseNode(id="wait"))
    .add_node(FunctionNode(
        id="check_order",
        function_id="validate_order",  # From: kapso functions list
        save_response_to="order"
    ))
    .add_node(DecideNode(
        id="order_status",
        conditions=[
            Condition(label="found", description="Order exists and is valid"),
            Condition(label="not_found", description="Order doesn't exist")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(SendTextNode(
        id="order_details",
        message="Order {{order.number}}: {{order.status}}. Arriving {{order.delivery_date}}"
    ))
    .add_node(SendTextNode(
        id="order_error",
        message="Order not found. Please double-check the number."
    ))

    .add_edge("start", "ask_order")
    .add_edge("ask_order", "wait")
    .add_edge("wait", "check_order", "response")
    .add_edge("check_order", "order_status")
    .add_edge("order_status", "order_details", "found")
    .add_edge("order_status", "order_error", "not_found")
)

StartNode(id="start")  # Always first, connects to everything

# Static message
SendTextNode(id="msg1", message="Hello!")

# AI message
SendTextNode(
    id="msg2",
    message=AIField(prompt="Generate a friendly greeting for {{customer_name}}"),
    provider_model_name="claude-3-5-sonnet"
)

# With variables
SendTextNode(id="msg3", message="Hi {{customer_name}}, order {{order_id}} is ready!")

SendTemplateNode(
    id="template1",
    template_id="order_confirmation",
    parameters={"1": "{{customer_name}}", "2": "{{order_total}}"}
)

from kapso.builder.flows.nodes.send_interactive import InteractiveButton

SendInteractiveNode(
    id="buttons1",
    header_text="Choose an option:",
    body_text="How can we help you?",
    buttons=[
        InteractiveButton(id="opt1", title="Support"),
        InteractiveButton(id="opt2", title="Sales"),
        InteractiveButton(id="opt3", title="Billing"),
    ]
)

from kapso.builder.flows.nodes.send_interactive import ListRow, ListSection

SendInteractiveNode(
    id="list1",
    header_text="Our Services",
    body_text="Select a service:",
    list_sections=[
        ListSection(title="Main Services", rows=[
            ListRow(id="tech", title="Technical Support"),
            ListRow(id="sales", title="Sales Inquiry")
        ])
    ]
)

WaitForResponseNode(id="wait1")  # Sets {{last_user_input}}

DecideNode(
    id="decide1",
    conditions=[
        Condition(label="yes", description="User agrees or says yes"),
        Condition(label="no", description="User declines or says no")
    ],
    provider_model_name="claude-3-5-sonnet"
)

AgentNode(
    id="agent1",
    system_prompt="You're a helpful customer service agent.",
    provider_model_name="claude-3-5-sonnet",
    max_iterations=10
)

from kapso.builder.flows.nodes.agent import FlowAgentMcpServer

# HTTP streamable transport only
mcp = FlowAgentMcpServer(
    name="Context Server",
    url="https://mcp.context7.ai/v1",
    description="Documentation access",
    headers={"Authorization": "Bearer token"}
)

AgentNode(
    id="agent1",
    system_prompt="Help with documentation questions.",
    provider_model_name="claude-3-5-sonnet",
    mcp_servers=[mcp]
)

from kapso.builder.flows.nodes.agent import FlowAgentWebhook

check_order = FlowAgentWebhook(
    name="check_order_status",
    url="https://api.example.com/orders/{{order_id}}",
    http_method="GET",
    description="Retrieve latest status for a given order"
)

AgentNode(
    id="agent_with_webhook",
    system_prompt="Help customers track orders and answer questions.",
    provider_model_name="claude-sonnet-4-20250514",
    webhooks=[check_order]
)

FunctionNode(
    id="func1",
    function_id="your_function_id",  # From: kapso functions list
    save_response_to="result"  # Saves output to {{result}}
)

HandoffNode(
    id="handoff1"
)

# Flow variables (most common)
"Hello {{customer_name}}"

# Context variables
"Your number is {{context.phone_number}}"
"Channel: {{context.channel}}"

# System variables
"Flow started at {{system.started_at}}"
"Flow ID: {{system.flow_id}}"

# Last user input (after wait node)
"You said: {{last_user_input}}"

# Function nodes
FunctionNode(save_response_to="my_variable")

# Webhook nodes
WebhookNode(save_response_to="api_response")

# Agent nodes (use save_variable tool)
# Agent can call: save_variable(key="user_email", value="user@example.com")

# 1. Create decide node
decide = DecideNode(
    id="classify",
    conditions=[
        Condition(label="urgent", description="Urgent or emergency"),
        Condition(label="normal", description="Standard inquiry")
    ],
    provider_model_name="claude-3-5-sonnet"
)

# 2. Add edges with MATCHING labels
.add_edge("classify", "urgent_handler", "urgent")
.add_edge("classify", "normal_handler", "normal")

Condition(label="interested", description="User shows interest, asks questions, or wants to proceed")
Condition(label="not_interested", description="User declines, says no, or seems uninterested")
Condition(label="unclear", description="User's intent is unclear or they need more information")

# Personalized greeting
AIField(prompt="Generate a warm, professional greeting for {{customer_name}} based on their previous order of {{last_product}}")

# Support response
AIField(prompt="Provide a helpful solution for this technical issue: {{last_user_input}}. Keep it concise and actionable.")

# Follow-up question
AIField(prompt="Ask a relevant follow-up question based on the user's response: {{last_user_input}}")

AIField(prompt='Generate order confirmation parameters as JSON: {"customer_name": "{{customer_name}}", "order_total": "calculate from {{items}}", "delivery_date": "estimate based on {{location}}"}')

# ❌ Wrong - labels don't match
Condition(label="yes", ...)
.add_edge("decide", "next_step", "affirmative")

# ✅ Correct - labels match exactly
Condition(label="yes", ...)
.add_edge("decide", "next_step", "yes")

# ❌ Wrong - no model specified
message=AIField(prompt="Generate greeting")

# ✅ Correct - model specified
message=AIField(prompt="Generate greeting")
provider_model_name="claude-3-5-sonnet"

# ❌ Wrong - phone_number is in context
"Your number: {{phone_number}}"

# ✅ Correct - use context namespace
"Your number: {{context.phone_number}}"

# ❌ Missing edges
.add_node(SendTextNode(id="greeting"))
.add_node(WaitForResponseNode(id="wait"))
# No edge between nodes!

# ✅ Connected properly
.add_edge("greeting", "wait")