Skip to main content
AI assistants reference for building WhatsApp flows with the Kapso Builder SDK.

Quickstart

pip install kapso
kapso login
kapso pull --project-id <id>       # Pull existing project
kapso flow init <name>             # Create new flow
kapso flow push <name>             # Deploy flow

Project structure

project/
├── kapso.yaml              # Project config
├── flows/                  # Flow definitions
│   └── flow_name/
│       ├── flow.py         # Flow code
│       └── metadata.yaml   # Flow metadata
└── functions/              # JavaScript functions
    └── function.js

Core imports

from kapso.builder.flows import Flow
from kapso.builder.flows.nodes import *
from kapso.builder.flows.nodes.agent import FlowAgentWebhook, FlowAgentMcpServer
from kapso.builder.flows.conditions import Condition
from kapso.builder.ai.field import AIField
Full import reference →

The working pattern

This pattern handles most use cases:
from kapso.builder.flows import Flow
from kapso.builder.flows.nodes import *
from kapso.builder.flows.conditions import Condition
from kapso.builder.ai.field import AIField

flow = (Flow(name="Customer Support")
    .add_node(StartNode(id="start"))
    .add_node(SendTextNode(
        id="greeting",
        message=AIField(prompt="Generate a friendly greeting"),
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(WaitForResponseNode(id="wait"))
    .add_node(DecideNode(
        id="route",
        conditions=[
            Condition(label="support", description="Technical support needed"),
            Condition(label="sales", description="Sales question")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="support_agent",
        system_prompt="You're a helpful technical support agent.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="sales_agent",
        system_prompt="You're a friendly sales agent.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    
    .add_edge(source="start", target="greeting")
    .add_edge(source="greeting", target="wait")
    .add_edge(source="wait", target="route", label="response")
    .add_edge(source="route", target="support_agent", label="support")
    .add_edge(source="route", target="sales_agent", label="sales")
)

Function-first pattern

Start flows with functions to check data and route intelligently:
from kapso.builder.flows import Flow
from kapso.builder.flows.nodes import *
from kapso.builder.flows.conditions import Condition

# User verification flow
flow = (Flow(name="Check User First")
    .add_node(StartNode(id="start"))
    .add_node(FunctionNode(
        id="verify_user",
        function_id="check_user_by_phone",  # From: kapso functions list
        save_response_to="user_data"
    ))
    .add_node(DecideNode(
        id="route_by_user",
        conditions=[
            Condition(label="existing", description="User exists in system"),
            Condition(label="new", description="New user, needs onboarding")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="existing_user_agent",
        system_prompt="Welcome back {{user_data.name}}! Access to account: {{user_data.account_type}}",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="onboarding_agent",
        system_prompt="Welcome new user! Help them get started with account setup.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    
    .add_edge("start", "verify_user")
    .add_edge("verify_user", "route_by_user")
    .add_edge("route_by_user", "existing_user_agent", "existing")
    .add_edge("route_by_user", "onboarding_agent", "new")
)
Flow architecture guide →

Node reference

StartNode

Entry point for every flow:
StartNode(id="start")  # Required first node

SendTextNode

Send text messages:
# Static message
SendTextNode(
    id="msg1",
    message="Hello {{customer_name}}"
)

# AI-generated message  
SendTextNode(
    id="msg2",
    message=AIField(prompt="Generate greeting for {{customer_name}}"),
    provider_model_name="claude-sonnet-4-20250514"
)
Send text documentation →

SendInteractiveNode

Send buttons or lists:
# Buttons
from kapso.builder.flows.nodes.send_interactive import InteractiveButton, ListRow, ListSection

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

# List
SendInteractiveNode(
    id="list", 
    header_text="Services",
    body_text="Select service:",
    list_sections=[
        ListSection(title="Main", rows=[
            ListRow(id="tech", title="Tech Support"),
            ListRow(id="sales", title="Sales")
        ])
    ]
)
Interactive messages documentation →

WaitForResponseNode

Wait for user input:
WaitForResponseNode(
    id="wait"
)
Wait node documentation →

DecideNode

AI-powered routing:
DecideNode(
    id="decide",
    conditions=[
        Condition(label="yes", description="User agrees"),
        Condition(label="no", description="User declines")
    ],
    provider_model_name="claude-sonnet-4-20250514"
)
Critical: Edge labels must match condition labels exactly. Decision routing documentation →

AgentNode

Conversational AI agent:
AgentNode(
    id="agent",
    system_prompt="You're a customer service agent.",
    provider_model_name="claude-sonnet-4-20250514",
    max_iterations=10
)

# Agent with MCP server (HTTP streamable transport)
mcp_server = FlowAgentMcpServer(
    name="Documentation Server",
    url="https://mcp.context7.ai/v1",
    description="Access documentation",
    headers={"Authorization": "Bearer token"}
)

AgentNode(
    id="agent",
    system_prompt="You're a documentation assistant.",
    provider_model_name="claude-sonnet-4-20250514",
    mcp_servers=[mcp_server]
)

Agent webhooks

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

inventory_lookup = FlowAgentWebhook(
    name="inventory_lookup",
    url="https://api.example.com/inventory/{{sku}}",
    http_method="GET",
    description="Fetch current inventory levels",
    headers={"Authorization": "Bearer {{api_token}}"}
)

AgentNode(
    id="agent_with_webhooks",
    system_prompt="Assist customers with product availability questions.",
    provider_model_name="claude-sonnet-4-20250514",
    webhooks=[inventory_lookup]
)

Webhook with JSON schema

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

create_ticket = FlowAgentWebhook(
    name="create_ticket",
    url="https://api.support.com/tickets",
    http_method="POST",
    description="Create a support ticket and return the ticket ID",
    headers={"Content-Type": "application/json"},
    body={
        "user_id": "{{context.phone_number}}",
        "subject": "{{vars.issue_subject}}",
        "details": "{{last_user_input}}"
    },
    body_schema=json.dumps({
        "type": "object",
        "properties": {
            "user_id": {"type": "string"},
            "subject": {"type": "string"},
            "details": {"type": "string"}
        },
        "required": ["user_id", "subject", "details"]
    }),
    jmespath_query="data.ticket_id"
)

AgentNode(
    id="ticket_agent",
    system_prompt="File support tickets based on the conversation and share the ticket ID with the customer.",
    provider_model_name="claude-sonnet-4-20250514",
    webhooks=[create_ticket]
)
body_schema guides the LLM when constructing request payloads. Use it to describe the JSON structure you need and keep body for fixed values or variables you want to pass without AI changes. In practice you typically choose one or the other. Only include both when you deliberately want to seed a constant field alongside schema-generated content.
Agent node documentation →

FunctionNode

Execute JavaScript functions:
FunctionNode(
    id="func",
    function_id="calculate_total",  # From: kapso functions list
    save_response_to="result"  # Saves to {{result}}
)
Common function patterns:
# Check user registration
FunctionNode(
    id="check_user",
    function_id="verify_user_phone",  # Your deployed function
    save_response_to="user_status"
)

# Load user profile
FunctionNode(
    id="get_profile", 
    function_id="fetch_user_data",  # Your deployed function
    save_response_to="profile"
)

# Validate order
FunctionNode(
    id="validate_order",
    function_id="check_order_exists",  # Your deployed function  
    save_response_to="order_info"
)
Function node documentation →

WebhookNode

HTTP API calls:
WebhookNode(
    id="api",
    url="https://api.example.com/data",
    method="POST",
    request_body='{"user": "{{customer_name}}"}',
    save_response_to="api_result"
)

HandoffNode

Transfer to human:
HandoffNode(
    id="handoff"
)
Handoff documentation →

Variables

See Variables & Context for a detailed overview of the vars, system, context, and metadata namespaces.

Reading variables

# Flow variables
"Hello {{customer_name}}"

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

# System variables  
"Started: {{system.started_at}}"

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

Writing variables

# Function/webhook nodes
save_response_to="variable_name"

# Agent nodes use save_variable tool
# Agent calls: save_variable(key="email", value="user@example.com")
Variables documentation →

AI fields

Use AIField for dynamic content:
# Three equivalent ways
AIField(prompt="Generate greeting")          # Recommended
AIField.prompt("Generate greeting")          
AIField("Generate greeting")                 

# With variables
AIField(prompt="Greet {{customer_name}} who ordered {{product}}")
Requirement: Must specify provider_model_name when using AIField. AI fields documentation →

Edges and routing

Connect nodes with edges:
# Basic connection (uses default "next" label)
.add_edge(source="start", target="greeting")

# With custom label
.add_edge(source="wait", target="decide", label="response")

# Decision routing (labels must match conditions)
.add_edge(source="decide", target="support_agent", label="support")  # Matches Condition(label="support")

# Positional arguments also work
.add_edge("start", "greeting")  # source, target
.add_edge("decide", "agent", "support")  # source, target, label
Edges documentation →

CLI commands

Project setup

kapso login                         # Authenticate
kapso pull --project-id <id>        # Pull project
kapso push                          # Push all changes

Flow development

kapso flow init <name>              # Create flow
kapso flow push <name>              # Deploy flow
kapso flow pull <name>              # Pull from cloud
kapso flow list                     # List flows

Functions

kapso functions push <file.js>      # Upload function
kapso functions list                # List functions
kapso functions pull <name>         # Download function
Full CLI reference →

Flow triggers

Flows start from:
  1. WhatsApp messages: User sends message, {{last_user_input}} available
  2. API calls: External trigger with custom variables
Triggers documentation →

Common mistakes

Missing provider_model_name

# ❌ Wrong
message=AIField(prompt="Generate text")

# ✅ Correct
message=AIField(prompt="Generate text")
provider_model_name="claude-sonnet-4-20250514"

Edge labels don’t match conditions

# ❌ Wrong
Condition(label="support", ...)
.add_edge("decide", "agent", "help")  # Different label

# ✅ Correct  
Condition(label="support", ...)
.add_edge("decide", "agent", "support")  # Matching label

Wrong variable namespace

# ❌ Wrong
"Your number: {{phone_number}}"

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

Testing and debugging

View execution details:
  • Test flows in web interface
  • Check Flow Events tab for step-by-step execution
  • Variables tab shows current values
  • Errors tab displays failures
Events documentation →

Complete example

from kapso.builder.flows import Flow
from kapso.builder.flows.nodes import *
from kapso.builder.flows.conditions import Condition

flow = (Flow(name="Lead Capture")
    .add_node(StartNode(id="start"))
    .add_node(SendTextNode(
        id="welcome",
        message="Welcome! What brings you here today?"
    ))
    .add_node(WaitForResponseNode(id="wait"))
    .add_node(DecideNode(
        id="qualify",
        conditions=[
            Condition(label="interested", description="Shows interest"),
            Condition(label="not_interested", description="Not interested")
        ],
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(AgentNode(
        id="sales_agent",
        system_prompt="Qualify leads and collect contact info.",
        provider_model_name="claude-sonnet-4-20250514"
    ))
    .add_node(SendTextNode(
        id="thanks",
        message="Thanks for your time!"
    ))
    
    .add_edge("start", "welcome")
    .add_edge("welcome", "wait") 
    .add_edge("wait", "qualify", "response")
    .add_edge("qualify", "sales_agent", "interested")
    .add_edge("qualify", "thanks", "not_interested")
)

flow.validate()
Complete flow reference →

JavaScript functions

Kapso supports two platforms for serverless functions:

Cloudflare Workers (default)

// functions/my-function.js
async function handler(request, env) {
  // Get request data
  const body = await request.json();
  
  // Access secrets (set in web app)
  const apiKey = env.API_KEY;
  const dbUrl = env.DATABASE_URL;
  
  // Use KV storage (optional)
  await env.KV.put('user:123', JSON.stringify(body));
  const cached = await env.KV.get('user:123', { type: 'json' });
  
  // Your business logic
  const result = {
    message: 'Hello from Cloudflare Workers',
    data: body,
    cached: cached
  };
  
  // Return JSON response
  return new Response(JSON.stringify(result), {
    headers: { 'Content-Type': 'application/json' }
  });
}

Supabase Functions (Deno)

// functions/my-function.js
Deno.serve(async (req) => {
  // Get request data
  const body = await req.json();
  
  // Access secrets
  const apiKey = Deno.env.get('API_KEY');
  const supabaseUrl = Deno.env.get('SUPABASE_URL');
  
  // Direct database access (if using Supabase)
  const { createClient } = await import('@supabase/supabase-js');
  const supabase = createClient(supabaseUrl, Deno.env.get('SUPABASE_KEY'));
  
  const { data } = await supabase
    .from('users')
    .select('*')
    .eq('phone', body.phone);
  
  // Your business logic
  const result = {
    message: 'Hello from Supabase Functions',
    users: data
  };
  
  return new Response(JSON.stringify(result), {
    headers: { 'Content-Type': 'application/json' }
  });
});

Key differences

Cloudflare Workers:
  • async function handler(request, env)
  • Secrets via env.SECRET_NAME
  • Built-in KV storage via env.KV
  • Global edge deployment
Supabase Functions:
  • Deno.serve(async (req) => {})
  • Secrets via Deno.env.get('SECRET_NAME')
  • Direct Supabase database access
  • Deno standard library

Deployment

# Deploy to Cloudflare Workers (default)
kapso functions push my-function.js

# Deploy to Supabase Functions
kapso functions push my-function.js --platform supabase
Secrets setup:
  1. Deploy function first
  2. Go to Kapso web app → Functions → Select function → Secrets tab
  3. Add secrets: API_KEY, DATABASE_URL, etc.
  4. Access via env.SECRET_NAME (Cloudflare) or Deno.env.get('SECRET_NAME') (Supabase)
Functions documentation →
I