twilio-taskrouter-routing

What this skill does

## Overview

TaskRouter is Twilio's skills-based routing engine. Instead of building custom queuing logic, you define Workers (agents), Task Queues (groups), and Workflows (routing rules). TaskRouter matches incoming tasks to the best available worker.

```
Incoming Task → Workflow (routing rules) → Task Queue (skill match) → Worker (agent)
                                                                        ↓
                                                                   Reservation
                                                                   (accept/reject)
```

**Common mistake:** Developers reinvent TaskRouter in custom Node.js — don't. If you're building skills-based routing, queue management, or agent assignment, use TaskRouter.

---

## Prerequisites

- Twilio account — see `twilio-account-setup`
- `TWILIO_ACCOUNT_SID` and `TWILIO_AUTH_TOKEN` — see `twilio-iam-auth-setup`
- SDK: `pip install twilio` / `npm install twilio`
- For voice routing: a Twilio phone number with webhook configured — see `twilio-voice-twiml`
- For AI escalation: ConversationRelay with escalation tools — see `twilio-voice-conversation-relay`

---

## Quickstart

**Step 1 — Create a Workspace**

A Workspace is the top-level container for all TaskRouter resources.

**Python**
```python
import os
from twilio.rest import Client

client = Client(os.environ["TWILIO_ACCOUNT_SID"], os.environ["TWILIO_AUTH_TOKEN"])

workspace = client.taskrouter.v1.workspaces.create(
    friendly_name="Support Center",
    event_callback_url="https://yourapp.com/taskrouter-events"
)

workspace_sid = workspace.sid  # WSxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
print(workspace_sid)
```

**Node.js**
```node
const twilio = require("twilio");
const client = twilio(process.env.TWILIO_ACCOUNT_SID, process.env.TWILIO_AUTH_TOKEN);

const workspace = await client.taskrouter.v1.workspaces.create({
    friendlyName: "Support Center",
    eventCallbackUrl: "https://yourapp.com/taskrouter-events",
});

const workspaceSid = workspace.sid;
```

**Step 2 — Create Activities (agent states)**

**Python**
```python
# Available — worker can receive tasks
available = client.taskrouter.v1.workspaces(workspace_sid).activities.create(
    friendly_name="Available", available=True
)

# Offline — worker cannot receive tasks
offline = client.taskrouter.v1.workspaces(workspace_sid).activities.create(
    friendly_name="Offline", available=False
)

# On a task — worker is busy
on_task = client.taskrouter.v1.workspaces(workspace_sid).activities.create(
    friendly_name="On Task", available=False
)
```

**Step 3 — Create Workers (agents)**

> **Security:** Always use `json.dumps()` (Python) or `JSON.stringify()` (Node.js) to construct attribute payloads. String interpolation is vulnerable to JSON injection.

**Python**
```python
worker = client.taskrouter.v1.workspaces(workspace_sid).workers.create(
    friendly_name="Alice",
    attributes='{"skills": ["billing", "technical"], "languages": ["en", "es"], "department": "support"}'
)
```

**Node.js**
```node
const worker = await client.taskrouter.v1.workspaces(workspaceSid).workers.create({
    friendlyName: "Alice",
    attributes: JSON.stringify({
        skills: ["billing", "technical"],
        languages: ["en", "es"],
        department: "support",
    }),
});
```

**Step 4 — Create Task Queues**

**Python**
```python
# Billing queue — matches workers with "billing" skill
billing_queue = client.taskrouter.v1.workspaces(workspace_sid).task_queues.create(
    friendly_name="Billing",
    target_workers='skills HAS "billing"'
)

# Technical queue
tech_queue = client.taskrouter.v1.workspaces(workspace_sid).task_queues.create(
    friendly_name="Technical",
    target_workers='skills HAS "technical"'
)

# Catch-all queue
default_queue = client.taskrouter.v1.workspaces(workspace_sid).task_queues.create(
    friendly_name="Default",
    target_workers='1==1'  # matches all workers
)
```

**Step 5 — Create a Workflow (routing rules)**

**Python**
```python
import json

workflow_config = {
    "task_routing": {
        "filters": [
            {
                "filter_friendly_name": "Billing",
                "expression": "department == 'billing'",
                "targets": [
                    {"queue": billing_queue.sid, "timeout": 120}
                ]
            },
            {
                "filter_friendly_name": "Technical",
                "expression": "department == 'technical'",
                "targets": [
                    {"queue": tech_queue.sid, "timeout": 120}
                ]
            }
        ],
        "default_filter": {
            "queue": default_queue.sid
        }
    }
}

workflow = client.taskrouter.v1.workspaces(workspace_sid).workflows.create(
    friendly_name="Support Routing",
    configuration=json.dumps(workflow_config),
    assignment_callback_url="https://yourapp.com/assignment"
)
```

**Step 6 — Create a Task (from an incoming call)**

**Python**
```python
task = client.taskrouter.v1.workspaces(workspace_sid).tasks.create(
    attributes='{"department": "billing", "caller": "+15558675310", "priority": 1}',
    workflow_sid=workflow.sid
)
```

**Step 7 — Handle the Assignment Callback**

When TaskRouter finds a matching worker, it POSTs to your `assignment_callback_url`:

**Python (Flask)**
```python
@app.route("/assignment", methods=["POST"])
def assignment():
    task_sid = request.form["TaskSid"]
    worker_sid = request.form["WorkerSid"]
    reservation_sid = request.form["ReservationSid"]

    # Option A: Dequeue to the worker's phone
    return jsonify({
        "instruction": "dequeue",
        "from": "+15551234567",  # your Twilio number
        "post_work_activity_sid": available_activity_sid
    })

    # Option B: Conference the caller and agent
    # return jsonify({
    #     "instruction": "conference",
    #     "from": "+15551234567",
    #     "post_work_activity_sid": available_activity_sid
    # })
```

**Node.js (Express)**
```node
app.post("/assignment", (req, res) => {
    res.json({
        instruction: "dequeue",
        from: "+15551234567",
        post_work_activity_sid: availableActivitySid,
    });
});
```

---

## Key Patterns

### Skills-Based Routing

Match tasks to workers based on attributes:

| Worker expression | Matches |
|-------------------|---------|
| `skills HAS "billing"` | Workers whose `skills` array contains "billing" |
| `languages HAS "es"` | Spanish-speaking workers |
| `department == "support"` | Workers in support department |
| `experience > 5` | Workers with 5+ years experience |
| `skills HAS "billing" AND languages HAS "es"` | Spanish-speaking billing agents |

### Priority Routing

Tasks with higher priority are assigned first:

```python
# VIP customer — priority 10 (higher = first)
task = client.taskrouter.v1.workspaces(workspace_sid).tasks.create(
    attributes='{"department": "billing", "priority": 10, "vip": true}',
    workflow_sid=workflow.sid,
    priority=10
)
```

### AI Agent Escalation

When an AI agent (via TAC) escalates to a human, create a TaskRouter task with the AI's context:

```python
# From your escalation webhook handler
def handle_escalation(escalation_data):
    task = client.taskrouter.v1.workspaces(workspace_sid).tasks.create(
        attributes=json.dumps({
            "department": escalation_data["reason_code"],
            "conversation_id": escalation_data["conversation_id"],
            "profile_id": escalation_data["profile_id"],
            "ai_summary": escalation_data["summary"],
            "priority": 5
        }),
        workflow_sid=workflow.sid
    )
```

The human agent receives the AI's conversation summary and customer profile.

### Workflow with Timeout Escalation

Route to specialized queue first, then overflow to general:

```python
workflow_config = {
    "task_routing": {
        "filters": [
            {
                "filter_friendly_name": "Billing Specialist First",
                "expression": "department ==