Skip to main content

Custom Integration Guide

Build custom integrations with WorkFlux API. Connect to proprietary systems, create custom workflows, and extend platform capabilities.

Custom Integration Overview

WorkFlux provides comprehensive APIs and webhooks for custom integrations. Whether you need to connect proprietary systems, create custom workflows, or extend platform capabilities, our API makes it possible.

Business Value: Custom integrations unlock automation opportunities that competitors can't match. Companies using custom integrations see 40% higher ROI by automating unique workflows.

API Overview

WorkFlux REST API provides:

• Full CRUD operations for agents and conversations

• Webhook subscriptions for real-time events

• Custom workflow triggers

• Data export and reporting endpoints

Base URL: `https://api.workflux.ai/v1`

API Versioning:

• Current version: v1

• Version specified in URL path

• Backward compatibility maintained for 12 months

• Deprecation notices sent 90 days in advance

Authentication

All API requests require OAuth 2.0 authentication. See our API Authentication guide for detailed setup instructions.

Quick Start:

```bash

# 1. Get access token

curl -X POST https://api.workflux.ai/v1/auth/token \

-H 'Content-Type: application/json' \

-d '{

"client_id": "your_client_id",

"client_secret": "your_client_secret",

"grant_type": "client_credentials"

}'

# 2. Use token in requests

curl -X GET https://api.workflux.ai/v1/agents \

-H 'Authorization: Bearer {access_token}'

```

Webhooks

Subscribe to real-time events:

• Conversation started/completed

• Agent responses

• Escalation events

• Integration status updates

Webhook Configuration:

```bash

# Create webhook subscription

POST /webhooks

Content-Type: application/json

{

"url": "https://your-app.com/webhooks/workflux",

"events": [

"conversation.started",

"conversation.completed",

"agent.response"

],

"secret": "your_webhook_secret"

}

# Response:

{

"id": "webhook-123",

"url": "https://your-app.com/webhooks/workflux",

"events": ["conversation.started", "conversation.completed"],

"status": "active",

"created_at": "2024-01-01T12:00:00Z"

}

```

Webhook Testing:

```bash

# Test webhook delivery

POST /webhooks/{webhook_id}/test

# This sends a test event to your endpoint

```

Data Mapping & Transformation

Field Mapping Configuration:

```json

{

"source_field": "customer_email",

"target_field": "email",

"transformation": "lowercase",

"required": true,

"validation": {

"type": "email",

"pattern": "^[\\w-\\.]+@([\\w-]+\\.)+[\\w-]{2,4}$"

}

}

```

Transformation Functions:

• `lowercase` - Convert to lowercase

• `uppercase` - Convert to uppercase

• `trim` - Remove whitespace

• `format_date` - Format date strings

• `format_phone` - Format phone numbers

• `concat` - Concatenate fields

• `split` - Split string into array

• `lookup` - Lookup value from mapping table

Example Mapping Configuration:

```json

{

"mappings": [

{

"source": "first_name",

"target": "FirstName",

"transformation": "trim"

},

{

"source": "email",

"target": "Email",

"transformation": "lowercase",

"validation": "email"

},

{

"source": ["first_name", "last_name"],

"target": "FullName",

"transformation": "concat",

"separator": " "

}

]

}

```

Testing Your Integration

Integration Testing Checklist:

1. Authentication Test: Verify OAuth token generation and refresh

2. Connection Test: Test API connectivity and response times

3. Data Sync Test: Verify data mapping and transformation

4. Webhook Test: Test webhook delivery and signature verification

5. Error Handling Test: Test error scenarios and retry logic

6. Rate Limit Test: Verify rate limit handling

Test Script Example:

```python

import pytest

import requests

BASE_URL = 'https://api.workflux.ai/v1'

def test_authentication():

response = requests.post(

f'{BASE_URL}/auth/token',

json={

'client_id': 'test_client_id',

'client_secret': 'test_secret',

'grant_type': 'client_credentials'

}

)

assert response.status_code == 200

assert 'access_token' in response.json()

def test_list_agents(access_token):

response = requests.get(

f'{BASE_URL}/agents',

headers={'Authorization': f'Bearer {access_token}'}

)

assert response.status_code == 200

assert 'data' in response.json()

```