Skip to content

Quick Start

Get up and running with the AI Flow API in minutes, using any programming language.

Prerequisites

  • A webhook endpoint that can receive HTTP POST requests
  • Ability to send HTTP responses with JSON
  • (Optional) WebSocket support for real-time integration

Step 1: Set Up Your Webhook Endpoint

Create an HTTP endpoint that receives POST requests. Here are examples in different languages:

Python (Flask)

python
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json

    if event['type'] == 'user_speak':
        # Respond with a speak action
        return jsonify({
            'type': 'speak',
            'session_id': event['session']['id'],
            'text': f"You said: {event['text']}"
        })

    return '', 204  # No response needed

if __name__ == '__main__':
    app.run(port=3000)

Node.js (Express)

javascript
const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhook', (req, res) => {
  const event = req.body;

  if (event.type === 'user_speak') {
    return res.json({
      type: 'speak',
      session_id: event.session.id,
      text: `You said: ${event.text}`
    });
  }

  res.status(204).send();
});

app.listen(3000, () => {
  console.log('Webhook server running on port 3000');
});

Go

go
package main

import (
    "encoding/json"
    "net/http"
)

type Event struct {
    Type    string `json:"type"`
    Text    string `json:"text,omitempty"`
    Session Session `json:"session"`
}

type Session struct {
    ID string `json:"id"`
}

func webhook(w http.ResponseWriter, r *http.Request) {
    var event Event
    json.NewDecoder(r.Body).Decode(&event)

    if event.Type == "user_speak" {
        action := map[string]interface{}{
            "type":       "speak",
            "session_id": event.Session.ID,
            "text":       "You said: " + event.Text,
        }
        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(action)
        return
    }

    w.WriteHeader(http.StatusNoContent)
}

func main() {
    http.HandleFunc("/webhook", webhook)
    http.ListenAndServe(":3000", nil)
}

Ruby (Sinatra)

ruby
require 'sinatra'
require 'json'

post '/webhook' do
  event = JSON.parse(request.body.read)

  if event['type'] == 'user_speak'
    return JSON.generate({
      type: 'speak',
      session_id: event['session']['id'],
      text: "You said: #{event['text']}"
    })
  end

  status 204
end

Step 2: Handle Session Start

Respond when a call begins:

python
@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json

    if event['type'] == 'session_start':
        return jsonify({
            'type': 'speak',
            'session_id': event['session']['id'],
            'text': 'Welcome! How can I help you today?'
        })

    if event['type'] == 'user_speak':
        return jsonify({
            'type': 'speak',
            'session_id': event['session']['id'],
            'text': f"You said: {event['text']}"
        })

    return '', 204

Step 3: Expose Your Endpoint

Make your endpoint accessible to the AI Flow service:

  1. Local Development: Use a tunneling service like ngrok:

    bash
    ngrok http 3000
  2. Production: Deploy to a public URL (AWS, Heroku, Railway, etc.)

  3. Configure: Add your webhook URL in the AI Flow dashboard

Step 4: Test Your Integration

  1. Make a test call to your configured phone number
  2. Speak something
  3. Check your server logs to see events
  4. Verify responses are working

Complete Example

Here's a complete example that handles all event types:

python
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json
    session_id = event['session']['id']

    if event['type'] == 'session_start':
        return jsonify({
            'type': 'speak',
            'session_id': session_id,
            'text': 'Hello! How can I help you today?'
        })

    elif event['type'] == 'user_speak':
        user_text = event['text'].lower()

        if 'goodbye' in user_text or 'bye' in user_text:
            return jsonify({
                'type': 'hangup',
                'session_id': session_id
            })

        return jsonify({
            'type': 'speak',
            'session_id': session_id,
            'text': f"You said: {event['text']}"
        })

    elif event['type'] == 'session_end':
        print(f"Session {session_id} ended")
        return '', 204

    return '', 204

if __name__ == '__main__':
    app.run(port=3000, debug=True)

Next Steps