Webhook Integration Guide

Learn how to integrate Enfuse Smart Avatar webhooks into your applications to receive real-time conversation data, sentiment analysis, and human intervention alerts.

Overview

Enfuse Smart Avatar sends comprehensive webhook payloads for every conversation event, providing rich analytics and conversation intelligence that can be integrated into CRM systems, support platforms, analytics dashboards, and custom applications.

What You'll Receive

Each webhook contains:

• Current Message: User transcript or AI response with sentiment analysis
• Complete Conversation History: Full conversation context with timestamps
• Conversation Analytics: Health metrics, sentiment trends, and resolution probability
• Human Intervention Alerts: Automatic escalation recommendations
• User Information: Collected user details (name, email) when available
• Formatted Transcript: Human-readable conversation summary

Webhook Payload Structure

Core Message Data

{
  "message": "Hello, I need help with my account",
  "type": "user",
  "user_id": "john.doe@company.com",
  "organization_id": "12345678-1234-1234-1234-123456789012",
  "project_id": "87654321-4321-4321-4321-210987654321",
  "avatar_profile_id": "abcd1234-5678-9012-3456-789012345678",
  "conversation_id": "conv_2024_01_15_xyz789",
  "timestamp": "2024-01-15T14:30:25.123Z"
}

Sentiment Analysis

{
  "sentiment": {
    "score": -0.2,
    "label": "negative",
    "confidence": 0.85,
    "emotions": ["frustration", "concern"],
    "urgency": "medium",
    "frustration_indicators": 2,
    "satisfaction_signals": 0,
    "escalation_risk": "medium",
    "keywords": ["problem", "urgent", "help"]
  }
}

Conversation Analytics

{
  "conversation_analytics": {
    "cumulative_sentiment": -0.1,
    "sentiment_trend": "declining",
    "frustration_indicators": 3,
    "satisfaction_signals": 1,
    "human_intervention_recommended": true,
    "conversation_health": "fair",
    "escalation_risk": "medium",
    "turn_count": 8,
    "user_messages": 4,
    "ai_messages": 4,
    "resolution_probability": 0.65,
    "dominant_emotions": ["frustration", "concern"],
    "urgency_level": "medium"
  }
}

Human Intervention Detection

{
  "human_intervention": {
    "needed": true,
    "reason": "Multiple frustration indicators detected",
    "urgency": "high",
    "pattern": "escalating_complaints"
  }
}

User Information Collection

{
  "user_info": {
    "name": "John Doe",
    "email": "john.doe@company.com",
    "collected": true,
    "declined": false
  }
}

Conversation History

{
  "conversation_history": [
    {
      "message": "Hello, I need help with my account",
      "type": "user",
      "timestamp": "2024-01-15T14:30:25.123Z",
      "sentiment_score": -0.2,
      "sentiment_label": "negative"
    },
    {
      "message": "I'd be happy to help you with your account. What specific issue are you experiencing?",
      "type": "ai",
      "timestamp": "2024-01-15T14:30:28.456Z",
      "sentiment_score": 0.5,
      "sentiment_label": "positive"
    }
  ]
}

Formatted Transcript

{
  "conversation_transcript": "USER [14:30]: Hello, I need help with my account\nAI [14:30]: I'd be happy to help you with your account. What specific issue are you experiencing?\nUSER [14:31]: I can't log in and I'm getting an error message"
}

Integration Examples

Node.js / Express

Basic Webhook Receiver

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

// Middleware for parsing webhook JSON
app.use(express.json({ limit: '10mb' }));

// Webhook endpoint
app.post('/webhook/conversation', (req, res) => {
    const payload = req.body;
    
    console.log(`New ${payload.type} message from ${payload.user_id}`);
    console.log(`Message: "${payload.message}"`);
    console.log(`Sentiment: ${payload.sentiment?.label} (${payload.sentiment?.score})`);
    
    // Check for human intervention needs
    if (payload.human_intervention?.needed) {
        console.log('🚨 HUMAN INTERVENTION NEEDED:', payload.human_intervention.reason);
        // Trigger escalation workflow
        escalateConversation(payload);
    }
    
    // Process conversation analytics
    if (payload.conversation_analytics?.human_intervention_recommended) {
        // Update CRM with conversation health status
        updateCRMConversationHealth(payload);
    }
    
    // Store conversation data
    storeConversationData(payload);
    
    res.json({ success: true, message: 'Webhook received' });
});

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

CRM Integration Example

// CRM integration with conversation intelligence
app.post('/webhook/conversation', async (req, res) => {
    const payload = req.body;
    
    try {
        // Extract key information
        const customerInfo = {
            email: payload.user_info?.email || payload.user_id,
            name: payload.user_info?.name,
            lastContact: payload.timestamp,
            sentiment: payload.sentiment?.label,
            sentimentScore: payload.sentiment?.score,
            conversationHealth: payload.conversation_analytics?.conversation_health,
            needsAttention: payload.human_intervention?.needed
        };
        
        // Update CRM contact record
        await updateCRMContact(customerInfo);
        
        // Create support ticket if escalation needed
        if (payload.human_intervention?.needed) {
            const ticket = {
                title: `Escalation: ${payload.human_intervention.reason}`,
                description: payload.conversation_transcript,
                priority: payload.human_intervention.urgency,
                customerEmail: payload.user_id,
                conversationId: payload.conversation_id,
                sentiment: payload.sentiment?.label,
                urgency: payload.human_intervention.urgency
            };
            
            await createSupportTicket(ticket);
        }
        
        // Track conversation metrics
        await trackConversationMetrics({
            conversationId: payload.conversation_id,
            turnCount: payload.conversation_analytics?.turn_count,
            resolutionProbability: payload.conversation_analytics?.resolution_probability,
            sentiment: payload.sentiment?.score,
            duration: payload.conversation_context?.duration_seconds
        });
        
        res.json({ success: true });
    } catch (error) {
        console.error('CRM integration error:', error);
        res.status(500).json({ error: 'Processing failed' });
    }
});

Real-time Dashboard with Socket.IO

const io = require('socket.io')(server);

app.post('/webhook/conversation', (req, res) => {
    const payload = req.body;
    
    // Broadcast to real-time dashboard
    io.emit('conversation-event', {
        conversationId: payload.conversation_id,
        type: payload.type,
        message: payload.message,
        sentiment: payload.sentiment,
        health: payload.conversation_analytics?.conversation_health,
        interventionNeeded: payload.human_intervention?.needed,
        userInfo: payload.user_info,
        timestamp: payload.timestamp
    });
    
    // Track key metrics for dashboard
    const metrics = {
        activeConversations: getActiveConversationCount(),
        avgSentiment: payload.conversation_analytics?.cumulative_sentiment,
        escalationRate: calculateEscalationRate(),
        resolutionRate: payload.conversation_analytics?.resolution_probability
    };
    
    io.emit('dashboard-metrics', metrics);
    
    res.json({ success: true });
});

Python / FastAPI

Basic Webhook Receiver

from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

class WebhookPayload(BaseModel):
    message: str
    type: str
    user_id: str
    conversation_id: str
    sentiment: Optional[Dict[str, Any]] = None
    conversation_analytics: Optional[Dict[str, Any]] = None
    human_intervention: Optional[Dict[str, Any]] = None
    user_info: Optional[Dict[str, Any]] = None
    conversation_history: Optional[List[Dict[str, Any]]] = None
    conversation_transcript: Optional[str] = None

@app.post("/webhook/conversation")
async def receive_webhook(payload: WebhookPayload):
    logger.info(f"New {payload.type} message from {payload.user_id}")
    logger.info(f"Message: '{payload.message}'")
    
    if payload.sentiment:
        logger.info(f"Sentiment: {payload.sentiment.get('label')} ({payload.sentiment.get('score')})")
    
    # Check for human intervention needs
    if payload.human_intervention and payload.human_intervention.get('needed'):
        logger.warning(f"🚨 HUMAN INTERVENTION NEEDED: {payload.human_intervention.get('reason')}")
        await escalate_conversation(payload)
    
    # Process conversation analytics
    if payload.conversation_analytics and payload.conversation_analytics.get('human_intervention_recommended'):
        await update_crm_conversation_health(payload)
    
    # Store conversation data
    await store_conversation_data(payload.dict())
    
    return {"success": True, "message": "Webhook received"}

Advanced CRM Integration

import asyncio
import aiohttp
from datetime import datetime

@app.post("/webhook/conversation")
async def advanced_webhook_handler(payload: WebhookPayload):
    try:
        # Extract customer information
        customer_data = {
            "email": payload.user_info.get("email") if payload.user_info else payload.user_id,
            "name": payload.user_info.get("name") if payload.user_info else None,
            "last_contact": payload.timestamp if hasattr(payload, 'timestamp') else datetime.now().isoformat(),
            "sentiment": payload.sentiment.get("label") if payload.sentiment else "neutral",
            "sentiment_score": payload.sentiment.get("score") if payload.sentiment else 0.0,
            "conversation_health": payload.conversation_analytics.get("conversation_health") if payload.conversation_analytics else "unknown",
            "needs_attention": payload.human_intervention.get("needed") if payload.human_intervention else False
        }
        
        # Parallel processing of multiple integrations
        tasks = [
            update_crm_contact(customer_data),
            track_conversation_metrics(payload),
            update_analytics_dashboard(payload)
        ]
        
        # Create support ticket if needed
        if payload.human_intervention and payload.human_intervention.get("needed"):
            ticket_data = {
                "title": f"Escalation: {payload.human_intervention.get('reason')}",
                "description": payload.conversation_transcript or "No transcript available",
                "priority": payload.human_intervention.get("urgency", "medium"),
                "customer_email": payload.user_id,
                "conversation_id": payload.conversation_id,
                "sentiment": payload.sentiment.get("label") if payload.sentiment else "neutral",
                "urgency": payload.human_intervention.get("urgency", "low")
            }
            tasks.append(create_support_ticket(ticket_data))
        
        # Execute all tasks concurrently
        await asyncio.gather(*tasks, return_exceptions=True)
        
        return {"success": True, "processed": len(tasks)}
        
    except Exception as e:
        logger.error(f"Webhook processing error: {e}")
        return {"success": False, "error": str(e)}

async def update_crm_contact(customer_data: dict):
    """Update CRM with customer conversation data"""
    # Implementation for your CRM (Salesforce, HubSpot, etc.)
    pass

async def create_support_ticket(ticket_data: dict):
    """Create support ticket for escalated conversations"""
    # Implementation for your ticketing system (Zendesk, Freshdesk, etc.)
    pass

async def track_conversation_metrics(payload: WebhookPayload):
    """Track conversation metrics for analytics"""
    metrics = {
        "conversation_id": payload.conversation_id,
        "turn_count": payload.conversation_analytics.get("turn_count") if payload.conversation_analytics else 0,
        "resolution_probability": payload.conversation_analytics.get("resolution_probability") if payload.conversation_analytics else 0.0,
        "sentiment_score": payload.sentiment.get("score") if payload.sentiment else 0.0,
        "escalated": payload.human_intervention.get("needed") if payload.human_intervention else False
    }
    # Store in your analytics database
    pass

WebSocket Real-time Updates

from fastapi import WebSocket, WebSocketDisconnect
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, data: dict):
        for connection in self.active_connections:
            try:
                await connection.send_json(data)
            except:
                self.active_connections.remove(connection)

manager = ConnectionManager()

@app.websocket("/ws/dashboard")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            await websocket.receive_text()  # Keep connection alive
    except WebSocketDisconnect:
        manager.disconnect(websocket)

@app.post("/webhook/conversation")
async def webhook_with_realtime(payload: WebhookPayload):
    # Process webhook normally
    await process_webhook(payload)
    
    # Broadcast to real-time dashboard
    dashboard_event = {
        "type": "conversation-event",
        "conversation_id": payload.conversation_id,
        "message_type": payload.type,
        "sentiment": payload.sentiment.get("label") if payload.sentiment else "neutral",
        "health": payload.conversation_analytics.get("conversation_health") if payload.conversation_analytics else "unknown",
        "intervention_needed": payload.human_intervention.get("needed") if payload.human_intervention else False,
        "user_email": payload.user_id,
        "timestamp": datetime.now().isoformat()
    }
    
    await manager.broadcast(dashboard_event)
    
    return {"success": True}

Best Practices

Security

• Verify webhook signatures if your platform provides them
• Use HTTPS endpoints for all webhook receivers
• Validate payload structure before processing
• Implement rate limiting to prevent abuse
• Store sensitive data securely (user info, conversation content)

Reliability

• Return HTTP 200 quickly to acknowledge receipt
• Process heavy operations asynchronously to avoid timeouts
• Implement retries for failed downstream operations
• Log all webhook events for debugging and auditing
• Monitor webhook performance and error rates

Data Handling

• Respect user privacy when storing conversation data
• Implement data retention policies for conversation history
• Anonymize sensitive information where possible
• Comply with GDPR/CCPA requirements for user data
• Provide data export/deletion capabilities

Error Handling

app.post('/webhook/conversation', async (req, res) => {
    try {
        // Process webhook
        await processWebhook(req.body);
        res.json({ success: true });
    } catch (error) {
        console.error('Webhook error:', error);
        
        // Return success to prevent retries for invalid data
        if (error.type === 'VALIDATION_ERROR') {
            res.json({ success: false, error: 'Invalid payload' });
        } else {
            // Return 500 for temporary issues to trigger retries
            res.status(500).json({ error: 'Processing failed' });
        }
    }
});

Testing Your Integration

Using the Sample Webhook Loggers

We provide complete webhook logger implementations that you can use for testing and as reference:

Node.js Logger

# Clone and run the Node.js webhook logger
cd webhook-logger
npm install
npm start

# Your webhook URL: http://localhost:8080/webhook
# View logs: http://localhost:8080/view

Python Logger

# Clone and run the Python webhook logger
cd webhook-logger-python
pip install -r requirements.txt
python main.py

# Your webhook URL: http://localhost:8080/webhook
# View logs: http://localhost:8080/view
# API docs: http://localhost:8080/docs

Testing Checklist

• Webhook endpoint returns HTTP 200 for valid payloads
• All required fields are processed correctly
• Human intervention alerts trigger appropriate actions
• Sentiment analysis data is captured and used
• User information is handled securely
• Conversation history is stored/processed as needed
• Error handling works for malformed payloads
• Performance is acceptable under load

Webhook Configuration

Configure your webhook URL in the Enfuse Smart Avatar dashboard:

1. Go to your avatar profile in the main interface
2. Scroll to "Conversation Webhook URL" in the profile settings
3. Enter your webhook URL (must be HTTPS in production)
4. Save your avatar profile to activate the webhook
5. Enable webhook delivery for your project

Your webhook will receive events for all conversations in that project.

Support

For integration support:

• Documentation: docs.oktalkto.me
• Sample Code: Check the webhook-logger and webhook-logger-python directories
• Technical Issues: Contact support through the dashboard

Changelog

• v2.0.0: Added conversation analytics, human intervention detection, and user information collection
• v1.1.0: Enhanced sentiment analysis with emotion detection and keywords
• v1.0.0: Initial webhook implementation with basic message events