From twilio-developer-kit
Route tasks to agents using Twilio TaskRouter. Covers Workers, Task Queues, Workflows, Reservations, skills-based routing, and common gotchas (hyphen attributes, HAS operator, reservation cascade). Use this skill for any multi-agent contact center, support queue, or AI agent escalation routing.
npx claudepluginhub twilio/ai --plugin twilio-developer-kitThis skill uses the workspace's default tool permissions.
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.
Guides Next.js Cache Components and Partial Prerendering (PPR): 'use cache' directives, cacheLife(), cacheTag(), revalidateTag() for caching, invalidation, static/dynamic optimization. Auto-activates on cacheComponents: true.
Guides building MCP servers enabling LLMs to interact with external services via tools. Covers best practices, TypeScript/Node (MCP SDK), Python (FastMCP).
Share bugs, ideas, or general feedback.
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.
twilio-account-setupTWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN — see twilio-iam-auth-setuppip install twilio / npm install twiliotwilio-voice-twimltwilio-voice-conversation-relayStep 1 — Create a Workspace
A Workspace is the top-level container for all TaskRouter resources.
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
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
# 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) orJSON.stringify()(Node.js) to construct attribute payloads. String interpolation is vulnerable to JSON injection.
Python
worker = client.taskrouter.v1.workspaces(workspace_sid).workers.create(
friendly_name="Alice",
attributes='{"skills": ["billing", "technical"], "languages": ["en", "es"], "department": "support"}'
)
Node.js
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
# 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
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
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)
@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)
app.post("/assignment", (req, res) => {
res.json({
instruction: "dequeue",
from: "+15551234567",
post_work_activity_sid: availableActivitySid,
});
});
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 |
Tasks with higher priority are assigned first:
# 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
)
When an AI agent (via TAC) escalates to a human, create a TaskRouter task with the AI's context:
# 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.
Route to specialized queue first, then overflow to general:
workflow_config = {
"task_routing": {
"filters": [
{
"filter_friendly_name": "Billing Specialist First",
"expression": "department == 'billing'",
"targets": [
{"queue": billing_queue.sid, "timeout": 60}, # Try billing queue for 60s
{"queue": default_queue.sid, "timeout": 120} # Overflow to general
]
}
],
"default_filter": {
"queue": default_queue.sid
}
}
}
# Set worker to available
client.taskrouter.v1.workspaces(workspace_sid) \
.workers(worker_sid) \
.update(activity_sid=available_activity_sid)
# Get real-time worker statistics
stats = client.taskrouter.v1.workspaces(workspace_sid) \
.workers \
.statistics() \
.fetch()
print(f"Available: {stats.realtime['total_available_workers']}")
| Agents | Architecture | Notes |
|---|---|---|
| < 10 | Single workflow, one queue per skill | No Flex needed — agents use phone |
| 10-50 | Multi-queue workflows, skills-based routing | Flex recommended for desktop |
| 50+ | Multi-tier workflows, priority routing, real-time monitoring | Full Flex + supervisor tools |
# WRONG — hyphens in attribute keys break workflow expressions
worker = client.taskrouter.v1.workspaces(workspace_sid).workers.create(
friendly_name="Alice",
attributes='{"skill-level": 5}' # hyphen breaks expression evaluation
)
# RIGHT — use underscores or camelCase
worker = client.taskrouter.v1.workspaces(workspace_sid).workers.create(
friendly_name="Alice",
attributes='{"skill_level": 5}'
)
No error — the expression silently fails to match.
# WRONG — "billing" is a string, not an array. HAS silently matches nothing.
target_workers = 'department HAS "billing"'
# RIGHT — use == for string attributes
target_workers = 'department == "billing"'
# RIGHT — use HAS only for arrays
target_workers = 'skills HAS "billing"' # skills: ["billing", "technical"]
Tasks sit in queue forever with no error.
When a reservation times out:
Fix: Set the timeout Activity to a short-duration state, not "Offline". Or implement a reservation timeout handler that keeps the worker available:
@app.route("/taskrouter-events", methods=["POST"])
def taskrouter_event():
event_type = request.form["EventType"]
if event_type == "reservation.timeout":
worker_sid = request.form["WorkerSid"]
# Keep worker available instead of moving to offline
client.taskrouter.v1.workspaces(workspace_sid) \
.workers(worker_sid) \
.update(activity_sid=available_activity_sid)
return "", 200
Updating an Activity's available flag returns 200 OK but may not change the value if workers are currently in that activity. Create new activities instead of modifying existing ones.
skill-level is treated as subtraction (skill minus level). Error 20001. Always use underscores: skill_level.HAS on non-array silently matches nothing — department HAS "billing" on a string attribute is accepted at creation but never matches. Tasks sit in queue forever with no error.available flag is silently immutable — Updating returns 200 OK but does not change the value. Must delete and recreate the Activity.multiTaskEnabled cannot be reverted to false — Once enabled on a Workspace, cannot be disabled. One-way door.default_filter as catch-all.friendlyName is case-insensitive unique — "alice" collides with "Alice".workflowSid is required for task creation — API does not auto-select a default Workflow.page query param not supported — Use PageToken for pagination. page returns error 40153.twilio-conference-callstwilio-call-recordingstwilio-voice-conversation-relaytwilio-voice-twiml