PipeCast Bridge — Webhook Call Transfer & Grand Estancia Bots¶

Status: Planning
Date: 2026-04-02
Feature Name: PipeCast Bridge (Pipecat + AstraPBX Call Transfer Bridge)

Problem¶

1. Secrets management doesn't scale. Managing 20+ client orgs with hardcoded API keys in pipecat bots means redeployment for every credential change.
2. Functions are dumb. flow_converter.py creates handlers that only transition nodes — no value mapping, no state, no logic. Any real behavior requires Python code, defeating the flow editor's purpose.
3. Disconnected systems. pipecat-flow (SQLite) and AstraPBX (MySQL/Sequelize) have separate databases. Org credentials must be duplicated.

Goal: Build and deploy bots entirely from the flow editor — functions, webhooks, value mappings, secrets — with zero Python code per client.

Architecture Decision: Shared DB, Same VPS¶

Decision: Deploy pipecat-flow on the same VPS as AstraPBX (89.116.31.109). Migrate pipecat-flow from SQLite to MySQL, sharing AstraPBX's MySQL instance.

┌─────────────────────────────────────────────────────┐
│  Cloud VPS (89.116.31.109)                          │
│                                                     │
│  ┌──────────┐   ┌──────────────┐   ┌────────────┐  │
│  │ AstraPBX │   │ pipecat-flow │   │  Asterisk   │  │
│  │ :3000    │   │ :7860        │   │  AMI/ARI    │  │
│  └────┬─────┘   └──────┬───────┘   └─────────────┘  │
│       │                │                             │
│       └──────┬─────────┘                             │
│              ▼                                       │
│     ┌────────────────┐                               │
│     │  MySQL (shared) │                              │
│     │  pbx_multitenant│                              │
│     │                 │                              │
│     │  organizations  │ ← AstraPBX owns              │
│     │  api_keys       │                              │
│     │  ...            │                              │
│     │                 │                              │
│     │  pipecat_bots   │ ← pipecat-flow owns          │
│     │  pipecat_keys   │                              │
│     └────────────────┘                               │
│                                                      │
│  ┌─────────────────┐                                 │
│  │ flow-editor     │ ← Static site / Cloudflare      │
│  │ (reads orgs via │   Pages or same VPS              │
│  │  pipecat API)   │                                  │
│  └─────────────────┘                                 │
└──────────────────────────────────────────────────────┘

External:
┌──────────────┐
│  LogsUpdate  │  ← GCP Cloud Run (events.astradial.com)
│  /bot-bridge │    JWT proxy + ticket creation
└──────────────┘

Why same VPS + shared MySQL: - Single source of truth — pipecat reads organizations table directly for api_key, api_secret, context_prefix. No sync, no duplication. - Flow editor can list orgs — admin API returns org names/IDs from same DB. When creating a bot, you pick the org. Bot↔DID↔Queue assignment all in one place. - Simpler ops — one server to SSH into, one DB to backup, one deploy pipeline. - Secure — DB not exposed over network. Both apps connect via localhost:3306. - Performance is NOT a concern — pipecat-flow is a lightweight WebSocket relay + state machine. All heavy compute (VAD, STT, TTS, LLM) runs on Gemini Live cloud. The VPS just shuffles PCM audio bytes and manages flow state — minimal CPU/RAM overhead on top of existing Asterisk + AstraPBX. - Call transfers via localhost — pipecat calls AstraPBX at localhost:3000 directly. No network hop, no JWT token dance, no secrets management. Internal auth only.

What stays external: LogsUpdate on GCP Cloud Run (events.astradial.com) — used only for Firestore operations (ticket creation, CRM logging). Not needed for call transfers.

Solution Architecture¶

Call Transfer Flow (direct localhost — Option A)¶

Pipecat Bot (webhook post_action)
    ↓ POST localhost:3000/api/v1/calls/{channel_id}/transfer
    ↓ Internal auth (shared secret or localhost whitelist)
AstraPBX API (same VPS)
    ↓ AMI Redirect
Asterisk → routes call to queue/extension/DID

No LogsUpdate proxy, no JWT, no secrets in flow JSON. Just a localhost HTTP call.

Ticket Creation Flow (via LogsUpdate)¶

Pipecat Bot (webhook post_action)
    ↓ POST events.astradial.com/bot-bridge {action: "create_ticket", ...}
LogsUpdate (GCP Cloud Run)
    ↓ Firestore write
Ticket created in /astrapbx/{org_id}/tickets/

Component Design¶

1. Enhanced Function System (flow_converter.py + flow editor)¶

The core change: Functions defined in the flow editor should be able to do everything — not just transition nodes. Add two new capabilities to the FlowFunction schema:

a) state_mappings — Set state values when function is called¶

{
  "name": "select_department",
  "description": "Route caller to the right department",
  "properties": {
    "department": {
      "type": "string",
      "enum": ["front_desk", "housekeeping", "restaurant", "spa", "maintenance", "billing"]
    }
  },
  "state_mappings": {
    "selected_department": "{args.department}",
    "department_queue": "{value_map.department_queues.{args.department}}"
  },
  "next_node_id": "confirm_transfer"
}

state_mappings lets the editor define what gets saved to flow_manager.state when the LLM calls a function. Template variables: - {args.xxx} — LLM-provided argument value - {value_map.xxx.yyy} — Lookup from a value map (see below) - {call.channel_id} — From call_metadata

b) value_maps — Key-value lookups defined at flow level¶

Added to the flow JSON root (alongside meta, nodes, edges):

{
  "meta": { "name": "Grand Estancia IVR" },
  "value_maps": {
    "department_queues": {
      "front_desk": "5001",
      "housekeeping": "5002",
      "restaurant": "5003",
      "spa": "5004",
      "maintenance": "5005",
      "billing": "5006"
    }
  },
  "nodes": [...]
}

Why: This replaces hardcoded Python dicts. The IVR bot maps department → queue number entirely in JSON. The flow editor provides a UI to edit these maps. Different clients get different maps without code changes.

c) flow_converter.py handler upgrade¶

Current handler (dumb):

async def handler(args: FlowArgs):
    return args, build_node_config(target_node_id)

New handler (smart):

async def handler(args: FlowArgs, flow_manager):
    # 1. Always store raw args in state
    flow_manager.state.update(args)

    # 2. Apply state_mappings with template resolution
    for key, template in state_mappings.items():
        flow_manager.state[key] = resolve_template(
            template, args=args, 
            value_maps=value_maps,
            call=flow_manager.call_metadata
        )

    # 3. Transition to next node
    return args, build_node_config(target_node_id)

2. Webhook Action Type (actions.py)¶

New built-in action "webhook". Template variables in body resolved from flow_manager.call_metadata and flow_manager.state.

Transfer (direct localhost):

{
  "type": "webhook",
  "url": "http://localhost:3000/api/v1/calls/{call.channel_id}/transfer",
  "auth": "internal",
  "body": {
    "destination": "{state.department_queue}",
    "destination_type": "queue"
  }
}

Ticket creation (via LogsUpdate):

{
  "type": "webhook",
  "url": "https://events.astradial.com/bot-bridge",
  "body": {
    "action": "create_ticket",
    "org_id": "{call.org_id}",
    "caller_number": "{call.endpoint}",
    "category": "{state.category}",
    "summary": "{state.summary}",
    "guest_name": "{state.guest_name}",
    "room_number": "{state.room_number}"
  }
}

Auth modes: - "auth": "internal" — adds internal shared secret header (X-Internal-Key) for localhost AstraPBX calls. Key stored in env var, not in flow JSON. - No auth flag — plain POST (for LogsUpdate or external webhooks) - No API keys ever appear in flow JSON.

3. Channel ID Plumbing (pipeline.py)¶

AstraPBX already sends channel_id, org_id, endpoint via WebSocket customParameters (ariClient.js:311-316). Extract and attach to flow_manager.call_metadata.

4. Database Migration (pipecat-flow: SQLite → MySQL)¶

Migrate pipecat-flow gateway from SQLite (aiosqlite) to MySQL (aiomysql or databases). Tables:

• pipecat_bots — bot definitions (flow JSON, model config)
• pipecat_api_keys — gateway API keys for WebSocket auth

Reads from AstraPBX's existing tables: - organizations — org_id, api_key, api_secret, context_prefix, name

5. Bot Bridge Endpoint (LogsUpdate)¶

POST /bot-bridge with two actions: - pbx_forward — JWT caching + proxy to AstraPBX (refactored from /whatsapp-bridge) - create_ticket — Firestore ticket creation

6. Transfer Endpoint Enhancement (AstraPBX)¶

Add destination_type parameter to transfer endpoint. Auto-resolves context prefix: - "extension" → ${org.context_prefix}_internal - "queue" → ${org.context_prefix}_queue - "external" → ${org.context_prefix}_outbound

7. Flow Editor Enhancements¶

a) Webhook action type — New option in action dropdown with URL, body, auth fields.

b) State Mappings UI — In the function inspector, below properties/required, add a "State Mappings" section. Key-value pairs where values support template syntax.

c) Value Maps UI — Flow-level panel (alongside meta/settings) to define named key-value lookup tables.

d) Org Selector — When the editor is connected to the gateway API, show org dropdown for bot assignment. The auth: "org" webhook flag resolves to whichever org the bot is assigned to.

8. Grand Estancia Bots¶

Both defined entirely as JSON flows in the editor. No Python code.

IVR Bot — Department Router¶

Flow: welcome → select_department (value_map lookup) → confirm_transfer (webhook post_action) → end

Runtime example — caller says "I want to book a table for dinner":

1. Call arrives → AudioSocket → pipecat
   flow_manager.call_metadata = {channel_id: "1712345678.123", org_id: "org_abc"}
   flow_manager.state = {}

2. Welcome node loads → Gemini gets system prompt + function schema

3. Caller says: "I want to book a table for dinner"

4. Gemini calls: select_department(department: "restaurant")

5. Handler runs (flow_converter.py):
   ├─ flow_manager.state.update({"department": "restaurant"})       # raw args
   ├─ state_mappings resolve:
   │   "selected_dept" → "{args.department}"             → "restaurant"
   │   "queue_number"  → "{value_map.department_queues.{args.department}}" → "5003"
   └─ flow_manager.state = {department: "restaurant", selected_dept: "restaurant", queue_number: "5003"}
   → Transitions to confirm_transfer node

6. Gemini says: "I'll connect you to our restaurant right away."

7. Post-actions fire:
   ├─ webhook: POST http://localhost:3000/api/v1/calls/1712345678.123/transfer
   │   Header: X-Internal-Key: <from env>
   │   Body: {"destination": "5003", "destination_type": "queue"}
   │   → AstraPBX does AMI Redirect → Asterisk routes to queue 5003
   └─ end_conversation: call handed off, bot exits

Full flow JSON:

{
  "meta": { "name": "Grand Estancia IVR", "version": "0.1.0" },
  "value_maps": {
    "department_queues": {
      "restaurant": "5003",
      "housekeeping": "5002",
      "reception": "5001"
    }
  },
  "nodes": [
    {
      "id": "welcome",
      "type": "initial",
      "position": { "x": 0, "y": 0 },
      "data": {
        "label": "Welcome",
        "role_messages": [
          {
            "role": "system",
            "content": "You are the Grand Estancia hotel receptionist. Listen to what the caller needs and route them to the right department: restaurant, housekeeping, or reception. Be warm and professional."
          }
        ],
        "functions": [
          {
            "name": "select_department",
            "description": "Route the caller to the appropriate department",
            "properties": {
              "department": {
                "type": "string",
                "enum": ["restaurant", "housekeeping", "reception"],
                "description": "The department the caller wants"
              }
            },
            "required": ["department"],
            "state_mappings": {
              "selected_dept": "{args.department}",
              "queue_number": "{value_map.department_queues.{args.department}}"
            },
            "next_node_id": "confirm_transfer"
          }
        ]
      }
    },
    {
      "id": "confirm_transfer",
      "type": "end",
      "position": { "x": 0, "y": 300 },
      "data": {
        "label": "Transfer Call",
        "task_messages": [
          {
            "role": "system",
            "content": "Tell the caller you are connecting them to their requested department now. Be brief and polite."
          }
        ],
        "post_actions": [
          {
            "type": "webhook",
            "url": "http://localhost:3000/api/v1/calls/{call.channel_id}/transfer",
            "auth": "internal",
            "body": {
              "destination": "{state.queue_number}",
              "destination_type": "queue"
            }
          },
          {
            "type": "end_conversation"
          }
        ]
      }
    }
  ],
  "edges": [
    { "id": "e1", "source": "welcome", "target": "confirm_transfer", "label": "select_department" }
  ]
}

Adding a new client with different departments? Just create a new flow with a different value_map and enum. Zero Python code. Zero redeployment.

Ticketing Bot — Missed Call Handler¶

Flow: welcome → identify_issue → FAQ path (end) OR collect_details → submit_ticket → confirmation → end

Triggered when: Queue timeout (no agent picks up). Configured as queue failover destination in AstraPBX dialplan.

Repos & File Changes¶

Implementation Order¶

1. DB migration — pipecat-flow SQLite → MySQL (shared with AstraPBX)
2. Channel ID plumbing — pipeline.py call_metadata extraction
3. Template engine — shared resolve_template() for state_mappings and webhook bodies
4. Enhanced flow_converter — state_mappings, value_map resolution in function handlers
5. Webhook action type — actions.py with template substitution and auth: "org"
6. Bot bridge endpoint — LogsUpdate /bot-bridge
7. Transfer endpoint — AstraPBX destination_type support
8. Flow editor schema — state_mappings, value_maps, webhook action fields
9. Flow editor UI — function state mappings, value maps panel, webhook action form
10. Grand Estancia IVR bot — JSON flow
11. Grand Estancia Ticketing bot — JSON flow

Test Cases¶

TC-1xx: Database & Shared Infrastructure¶

TC-2xx: Channel ID Plumbing¶

TC-3xx: Template Engine¶

TC-4xx: Enhanced Function Handlers (flow_converter)¶

TC-5xx: Webhook Action Type¶

TC-6xx: Bot Bridge Endpoint (LogsUpdate)¶

TC-7xx: Transfer Endpoint Enhancement (AstraPBX)¶

TC-8xx: Flow Editor — Schema & UI¶

TC-9xx: Grand Estancia IVR Bot¶

TC-10xx: Grand Estancia Ticketing Bot¶

TC-E2E: End-to-End Integration¶

Total: 84 test cases

Bugs / Fixes Log¶

Track issues found during implementation here.

Context Notes¶

• AstraPBX DB: MySQL (pbx_multitenant), Sequelize ORM, DB_HOST=localhost, DB_PORT=3306
• Auth decision: Option A (direct localhost call). For call transfers, pipecat calls AstraPBX API directly on localhost:3000 — no JWT, no LogsUpdate proxy. Internal auth via shared secret or localhost whitelist. LogsUpdate /bot-bridge used only for Firestore operations (ticket creation).
• channel_id comes from ARI channel.id, format like 1712345678.123
• channel.name (endpoint) format like PJSIP/101-00000001
• JWT tokens cached 23h in LogsUpdate (24h expiry from AstraPBX)
• AudioSocket: raw 16-bit signed linear PCM at 8kHz
• Flow converter passes pre_actions/post_actions through as-is from JSON
• AstraPBX context prefix: org_{timestamp}_{type} (e.g., org_m1abc_internal)
• Decision routing in editor uses action (Python code) + conditions — this still works alongside state_mappings