​
Sample Integrations

This page provides sample code for common operations with the Kapso API.

​
Starting an Agent Execution (Node.js)

const axios = require('axios');

async function startAgentExecution(apiKey, agentId, phoneNumber, initialMessage = null) {
  try {
    const response = await axios.post(
      `https://api.kapso.io/api/v1/agents/${agentId}/executions`,
      {
        phone_number: phoneNumber,
        message: initialMessage
      },
      {
        headers: {
          'X-API-Key': apiKey,
          'Content-Type': 'application/json'
        }
      }
    );
    
    console.log('Execution started:', response.data);
    // Example output:
    // Execution started: {
    //   "id": "fedcba98-7654-3210-fedc-ba9876543210",
    //   "status": "started",
    //   "started_at": "2023-10-22T10:15:00Z",
    //   "ended_at": null,
    //   "created_at": "2023-10-22T10:15:00Z",
    //   "updated_at": "2023-10-22T10:15:00Z"
    // }
    return response.data;
  } catch (error) {
    console.error('Error starting execution:', error.response?.data || error.message);
    throw error;
  }
}

// Example usage
startAgentExecution(
  'your_api_key_here',
  '01234567-89ab-cdef-0123-456789abcdef',
  '+1234567890',
  'I need help with my order'
);

​
Retrieving Conversation History (Python)

import requests

def get_conversation_messages(api_key, conversation_id):
    headers = {
        'X-API-Key': api_key,
        'Content-Type': 'application/json'
    }
    
    url = f"https://api.kapso.io/api/v1/whatsapp_conversations/{conversation_id}/whatsapp_messages"
    
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # Raise exception for 4XX/5XX responses
        
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error retrieving conversation messages: {str(e)}")
        raise

# Example usage
api_key = "your_api_key_here"
conversation_id = "conv-uuid-12345"
messages = get_conversation_messages(api_key, conversation_id)

for message in messages:
    direction = "→" if message['direction'] == 'outbound' else "←"
    print(f"{message['created_at']} {direction} {message['content']}")

​
Best Practices

​
Error Handling

  1. Implement exponential backoff for retrying failed requests
  2. Cache frequently accessed resources to reduce API load
  3. Monitor rate limits and adjust request timing accordingly

​
Security

  1. Rotate API keys regularly
  2. Use HTTPS for all API requests
  3. Validate webhook requests by checking headers and payload signatures
  4. Implement IP whitelisting for production environments

​
Performance

  1. Paginate large result sets
  2. Minimize the frequency of polling for updates
  3. Use webhook notifications instead of polling when possible