Chapter 6: Tasks and Task Groups
In the previous chapters you built an agent that can listen, speak, and call tools. That is enough for a simple Q&A bot. But real voice workflows are multi-step. You need to collect a business name, then services, then contact info, then read everything back for confirmation. Each step has its own tools, its own validation logic, and its own completion criteria. If you try to cram all of that into a single agent’s instruction set and tool list, you end up with a prompt that is thousands of tokens long, tools that conflict with each other, and an LLM that forgets what it was doing halfway through.
Tasks solve this. A task is a focused, reusable unit that takes control of the voice session, pursues a single objective, and returns a typed result when it is done. Think of it as a sub-conversation with a clear end state. The agent hands off to the task, the task runs its own tools and instructions, and when it calls complete(), control returns to the agent with structured data in hand.
The rest of this chapter walks through tasks, task groups, and three patterns that emerged from shipping a production voice agent: the two-step confirmation, optional field skip paths, and the readback gate.
What Tasks Are
A task is a Python class that extends AgentTask[T], where T is the type of result it produces. When a task starts, it replaces the active agent’s instructions and tools with its own. The LLM only sees the task’s context, which keeps it focused. When the task calls self.complete(result), the result is returned to whoever awaited the task, and the previous context is restored.
Key properties of tasks:
- Focused: A task has one job. Collect a phone number. Get consent. Confirm details.
- Typed: The result type is declared via generics. The caller gets back a dataclass, not a raw string.
- Reusable: The same task class can be used in different agents and workflows.
- Self-contained: Each task defines its own instructions, tools, and completion logic.
Defining an AgentTask
Every task follows the same structure: a result dataclass, a class that extends AgentTask, and a set of tool methods that drive the conversation toward completion.
The result type
Start by defining what the task produces. Use a dataclass:
from dataclasses import dataclass
@dataclass
class PhoneNumberResult:
phone_number: str
country_code: strThis is the contract between the task and its caller. When the task completes, the caller gets a PhoneNumberResult with exactly these fields.
The task class
Extend AgentTask with your result type as the generic parameter:
from livekit.agents import AgentTask, function_tool
from livekit.agents import RunContext
class CollectPhoneNumberTask(AgentTask[PhoneNumberResult]):
def __init__(self) -> None:
super().__init__(
instructions=(
"Collect the caller's phone number including country code. "
"Ask them to say or spell out their number. "
"Read it back to confirm before recording."
)
)The instructions parameter is the system prompt for this task. It replaces the parent agent’s instructions while the task is active.
on_enter: starting the interaction
The on_enter method is called when the task takes control. Use it to generate the first message:
async def on_enter(self) -> None:
self.session.generate_reply(
instructions="Ask the caller for their phone number, including country code."
)Note that we do not await the generate_reply call here. This is intentional. If on_enter is triggered from within a tool’s execution, awaiting the speech playout can cause a circular wait: the tool call remains active until on_enter returns, but playout cannot finish until the tool call resolves. Fire and forget.
Function tools
Define the tools available during this task using the @function_tool decorator. These are the only tools the LLM can call while this task is active:
@function_tool()
async def record_phone_number(
self,
ctx: RunContext,
phone_number: str,
country_code: str,
) -> str:
"""Record the caller's phone number.
Args:
phone_number: The phone number digits
country_code: The country code (e.g. +1, +44)
"""
# Validate
if len(phone_number) < 7:
raise ToolError("Phone number too short. Ask the caller to repeat it.")
return (
f"Recorded: {country_code} {phone_number}. "
"Read this back to the caller and ask if it's correct."
)The docstring and Args block are critical. They are what the LLM sees when deciding which tool to call and what arguments to pass.
Completing the task
Call self.complete(result) to end the task and return data to the caller:
@function_tool()
async def confirm_phone_number(self, ctx: RunContext) -> None:
"""Confirm the phone number after the caller validates it."""
self.complete(PhoneNumberResult(
phone_number=self._phone_number,
country_code=self._country_code,
))Once complete() is called, the task is done. Control returns to the agent or task group that started it.
Complete Example: Collecting a Phone Number
Here is the full task, from dataclass to completion:
import re
from dataclasses import dataclass
from livekit.agents import AgentTask, function_tool
from livekit.agents.llm import ToolError
from livekit.agents import RunContext
_PHONE_REGEX = r"^\+?[\d\s\-().]{7,20}$"
@dataclass
class PhoneNumberResult:
phone_number: str
country_code: str
class CollectPhoneNumberTask(AgentTask[PhoneNumberResult]):
def __init__(self) -> None:
super().__init__(
instructions=(
"Collect the caller's phone number with country code. "
"Read it back to confirm before finalizing. "
"Do NOT call confirm_phone_number until the caller "
"has explicitly said the number is correct."
),
)
self._phone_number: str | None = None
self._country_code: str | None = None
self._record_handle: object | None = None
async def on_enter(self) -> None:
self.session.generate_reply(
instructions="Ask the caller for their phone number including country code."
)
@function_tool()
async def record_phone_number(
self,
ctx: RunContext,
phone_number: str,
country_code: str,
) -> str:
"""Record the caller's phone number.
Args:
phone_number: The phone number digits
country_code: The country code (e.g. +1, +44)
"""
if self.done():
return ""
full_number = f"{country_code}{phone_number}"
if not re.match(_PHONE_REGEX, full_number):
raise ToolError(
f"Phone number looks invalid: {full_number}. "
"Ask the caller to repeat it."
)
self._phone_number = phone_number
self._country_code = country_code
self._record_handle = ctx.speech_handle
return (
f"Recorded: {country_code} {phone_number}. "
"Read this back to the caller and ask if it's correct. "
"Do NOT call confirm_phone_number yet."
)
@function_tool()
async def confirm_phone_number(self, ctx: RunContext) -> None:
"""Confirm the phone number after the caller validates it.
Call this only AFTER the caller has explicitly confirmed.
"""
if self.done():
return
if not self._phone_number:
raise ToolError(
"No phone number recorded yet. Call record_phone_number first."
)
if ctx.speech_handle == self._record_handle:
raise ToolError(
"Wait for the caller to confirm before proceeding. "
"Do not call this in the same turn as record_phone_number."
)
self.complete(PhoneNumberResult(
phone_number=self._phone_number,
country_code=self._country_code,
))The _record_handle check in confirm_phone_number is the two-step confirmation pattern. We will cover it in detail shortly.
TaskGroup: Sequential Execution
A single task collects one piece of information. A real onboarding flow collects five or six. TaskGroup chains tasks together in sequence, sharing conversation context across all of them.
Creating a task group
from livekit.agents.beta.workflows import TaskGroup
# Inside your agent or tool
collect_group = TaskGroup(chat_ctx=self.chat_ctx)
collect_group.add(
lambda: CollectBusinessIdentityTask(),
id="identity",
description="Business name and trade type",
)
collect_group.add(
lambda: CollectOperationsTask(),
id="operations",
description="Services, hours, and location",
)
collect_group.add(
lambda: CollectOwnerContactTask(),
id="contact",
description="Owner name and contact info",
)
results = await collect_group
task_results = results.task_results
# task_results["identity"] -> BusinessIdentityResult
# task_results["operations"] -> OperationsResult
# task_results["contact"] -> OwnerContactResultThe add() method takes three arguments:
- task_factory: A callable that returns a task instance. This is always a
lambdaor function, not a task instance directly. The factory pattern is important because it allows the framework to re-create the task if the user regresses to an earlier step. - id: A string key used to look up results in
task_results. - description: A human-readable description that helps the LLM understand the purpose of each step. This is used when the LLM decides whether to regress to an earlier task.
Why factories
You might wonder why we pass lambda: CollectBusinessIdentityTask() instead of just CollectBusinessIdentityTask(). The reason is regression. If a caller says “actually, let me change my business name” during the contact collection step, the TaskGroup needs to go back to the identity task. But that task has already completed and its internal state reflects the old data. The factory allows the framework to create a fresh instance of the task, starting clean.
Configuration
TaskGroup accepts several configuration options:
task_group = TaskGroup(
chat_ctx=self.chat_ctx, # Share conversation history
summarize_chat_ctx=True, # Summarize task interactions back to main context
return_exceptions=False, # Propagate exceptions immediately
on_task_completed=my_callback, # Called after each task completes
)chat_ctx: Pass the current agent’s chat context so tasks can see what was said before they started. Without this, each task begins with an empty conversation.summarize_chat_ctx: WhenTrue(the default), the framework summarizes the multi-turn exchanges within the task group into a single message and merges it back into the main context. This prevents the chat context from growing unboundedly.return_exceptions: WhenFalse(the default), an unhandled exception in any task stops the entire group. WhenTrue, the exception is stored intask_resultsand the group continues to the next task.on_task_completed: An async callback invoked after each sub-task finishes. Useful for updating progress indicators or logging.
The Two-Step Confirmation Pattern
This is the single most important pattern in this chapter. It solves a problem that every voice agent encounters in production.
The problem
When the LLM processes a tool call like record_business_name(name="Mike's Plumbing", trade="plumber"), it returns the tool result and immediately decides what to do next. A well-intentioned LLM will often do this:
- Call
record_business_namewith the caller’s data - See the tool result: “Recorded. Read this back to confirm.”
- Generate speech: “I have Mike’s Plumbing, a plumber. Is that correct?”
- Immediately call
confirm_business_namein the same response
Steps 3 and 4 happen in the same LLM turn. The agent asks for confirmation and confirms, all before the caller has said a word. The speech has not even started playing yet.
Prompt instructions alone do not reliably prevent this. You can write “Do NOT call confirm until the caller responds” in bold, uppercase, with exclamation marks. The LLM will still do it some percentage of the time.
The solution: speech_handle guard
Every tool call in LiveKit receives a RunContext that includes a speech_handle, an object representing the current speech generation. If two tool calls happen in the same LLM turn, they share the same speech_handle. If they happen in different turns (meaning the caller spoke in between), they have different handles.
This gives us a mechanical guarantee that the caller actually responded:
@function_tool()
async def record_business_name_and_trade(
self,
ctx: RunContext,
business_name: str,
trade_type: str,
) -> str:
"""Record the business name and trade type."""
# ... validation ...
self._business_name = business_name
self._trade_type = trade_type
self._record_handle = ctx.speech_handle # Save the handle
return (
f"Business name: {business_name}, Trade: {trade_type}. "
"Read these back to the caller and ask if that's correct. "
"Do NOT call confirm_business_identity yet - wait for the caller to respond."
)
@function_tool()
async def confirm_business_identity(self, ctx: RunContext) -> None:
"""Confirm the business name and trade after the caller validates them."""
if not self._business_name or not self._trade_type:
raise ToolError(
"No business identity recorded yet. "
"Call record_business_name_and_trade first."
)
if ctx.speech_handle == self._record_handle:
raise ToolError(
"Wait for the caller to confirm before proceeding. "
"Do not call this in the same turn as record_business_name_and_trade."
)
self.complete(BusinessIdentityResult(
business_name=self._business_name,
trade_type=self._trade_type,
))The key line is ctx.speech_handle == self._record_handle. If the handles match, both tool calls are happening in the same LLM generation, meaning the caller has not spoken yet. The ToolError is returned to the LLM, which forces it to generate speech (the readback) and wait for the next turn.
This pattern is not optional. Without it, your agent will skip user confirmation on roughly 10-20% of calls, depending on the LLM and prompt length. With it, the skip rate drops to zero.
The full flow
Here is what happens at runtime:
- Caller says: “It’s Mike’s Plumbing, I’m a plumber”
- LLM calls
record_business_name_and_trade(business_name="Mike's Plumbing", trade_type="plumber") - Tool returns: “Read these back… Do NOT call confirm yet”
- LLM tries to call
confirm_business_identityin the same turn ctx.speech_handle == self._record_handleisTrue, soToolErroris raised- LLM generates speech: “I have Mike’s Plumbing, a plumber. Is that correct?”
- Caller says: “Yes, that’s right”
- New LLM turn, new
speech_handle - LLM calls
confirm_business_identity(handles differ) and the task completes
Optional Fields with Skip Paths
Not every field in your workflow is required. Operating hours, for example, might be something a caller does not want to provide. You need a clean path for the task to complete without that data.
The pattern is straightforward: add a skip_X() tool that completes the task (or marks the field as skipped) without requiring input:
class CollectOperationsTask(AgentTask[OperationsResult]):
def __init__(self) -> None:
super().__init__(
instructions=(
"Collect the business's services and location (both required). "
"Also ask for operating hours, but if the caller declines "
"or doesn't have set hours, call skip_hours immediately. "
"Do NOT ask more than once if they decline."
)
)
self._services: list[str] | None = None
self._hours: str | None = None
self._location: str | None = None
def _check_completion(self) -> str | None:
if self._services is not None and self._location is not None:
self.complete(OperationsResult(
services=self._services,
hours=self._hours, # May be None
location=self._location,
))
return None
missing = []
if self._services is None:
missing.append("services offered")
if self._location is None:
missing.append("service area")
return f"Recorded. Still need: {', '.join(missing)}."
@function_tool()
async def record_hours(self, ctx: RunContext, hours: str) -> str | None:
"""Record the operating hours of the business."""
if self.done():
return None
self._hours = hours.strip()
return self._check_completion()
@function_tool()
async def skip_hours(self, ctx: RunContext) -> str | None:
"""Skip operating hours if the caller declines or doesn't want to provide them."""
if self.done():
return None
# self._hours remains None
return self._check_completion()
@function_tool()
async def record_services(self, ctx: RunContext, services: list[str]) -> str | None:
"""Record the list of services the business offers."""
# ... validation and recording ...
@function_tool()
async def record_location(self, ctx: RunContext, location: str) -> str | None:
"""Record the service area or location."""
# ... validation and recording ...The key points:
- Required fields have no skip tool. Services and location must be collected. There is no
skip_services(). The LLM has no way to bypass them. - The skip tool is explicit in the instructions. “If the caller declines, call
skip_hoursimmediately.” The LLM needs clear permission to use the skip path. - “Do NOT ask more than once.” Without this instruction, the LLM will often circle back and re-ask for optional fields. Callers find this irritating.
- The result type uses
Nonefor skipped fields.hours: str | Nonein the dataclass. Downstream code must handle theNonecase.
The Readback Gate Pattern
The two-step confirmation prevents the LLM from confirming without the caller’s input. The readback gate goes one step further: it prevents the LLM from asking for confirmation without first reading back all the details.
The problem
After collecting five or six fields across multiple tasks, you need the agent to read everything back to the caller before asking “Does that all look correct?” But LLMs take shortcuts. Given a summary of collected data and a confirm_details tool, the LLM might generate “Everything looks good, confirming now” and call confirm_details(confirmed=True) without ever reading the data aloud.
The solution: a readback flag
Add a readback_complete() tool that must be called before confirm_details() is allowed. Combine this with the speech_handle guard to ensure the readback actually played out:
class ConfirmDetailsTask(AgentTask[BusinessProfileResult]):
def __init__(self, profile: BusinessProfileResult) -> None:
self._profile = profile
self._readback_done: bool = False
self._readback_handle: object | None = None
summary = self._build_summary()
super().__init__(
instructions=(
"Read back ALL of the following details to the caller, "
"then call readback_complete. After the caller responds, "
"call confirm_details.\n\n"
f"{summary}"
),
)
async def on_enter(self) -> None:
summary = self._build_summary()
self.session.generate_reply(
instructions=f"Read back these details to the caller:\n{summary}",
allow_interruptions=False,
)
@function_tool()
async def readback_complete(self, ctx: RunContext) -> str:
"""Call this AFTER you have read all business details back to the caller."""
self._readback_done = True
self._readback_handle = ctx.speech_handle
return "Good. Now wait for the caller to confirm or request changes."
@function_tool()
async def confirm_details(self, ctx: RunContext, confirmed: bool) -> str | None:
"""Confirm or reject the business details.
Only call this AFTER readback_complete and the caller has responded.
Args:
confirmed: True if the caller confirms, False otherwise.
"""
if not self._readback_done:
raise ToolError(
"You must read back all details first. "
"Call readback_complete after reading them to the caller."
)
if ctx.speech_handle == self._readback_handle:
raise ToolError(
"Wait for the caller to respond before confirming. "
"Do not confirm in the same turn as the readback."
)
if confirmed:
self.complete(self._profile)
return None
return "Ask the caller which detail they'd like to change."There are three layers of defense here:
allow_interruptions=Falseon the readback speech. This ensures the agent finishes reading all details even if the caller says “uh-huh” partway through._readback_doneflag. The LLM cannot callconfirm_detailsuntilreadback_completehas been called. This is a hard gate, not a suggestion.speech_handleguard on the readback. Even afterreadback_completeis called, the LLM cannot confirm in the same turn. The caller must speak first.
Handling updates after readback
If the caller says “Actually, change the business name,” you need to allow updates and then require a fresh readback:
@function_tool()
async def update_field(self, ctx: RunContext, field_name: str, new_value: str) -> str:
"""Update a single business detail field."""
# ... validation ...
setattr(self._profile, field_name, new_value)
# Reset readback state -- caller must re-confirm after changes
self._readback_done = False
self._readback_handle = None
return f"Updated {field_name}. Read back the updated summary and ask to confirm again."Resetting _readback_done forces the entire readback-confirm cycle to repeat after any change. This prevents the agent from confirming stale data.
Prebuilt Tasks
LiveKit provides several ready-made tasks for common voice collection scenarios. These handle the tricky edge cases so you do not have to.
GetEmailTask
Collecting email addresses over voice is notoriously difficult. Callers say “john dot doe at gmail dot com” and STT transcribes it as “[email protected]” if you are lucky, or “john dot doe at g mail dot com” if you are not. GetEmailTask handles normalization of spoken patterns, converting words like “dot,” “underscore,” and “at” into their symbol equivalents:
from livekit.agents.beta.workflows import GetEmailTask
email_result = await GetEmailTask(chat_ctx=self.chat_ctx)
print(email_result.email_address) # "[email protected]"If the caller declines to provide an email, the task calls its built-in decline_email_capture() tool. You can customize this behavior with extra_instructions and additional tools.
GetAddressTask
Collects and validates a complete mailing address with international format support. Handles spoken address quirks like “dash” becoming - and “apartment” normalizing to a structured unit field:
from livekit.agents.beta.workflows import GetAddressTask
address_result = await GetAddressTask(chat_ctx=self.chat_ctx)GetDtmfTask
Collects keypad (DTMF) or spoken digits from callers. Essential for IVR flows where callers press keys on their phone or speak numbers aloud:
from livekit.agents.beta.workflows import GetDtmfTask
dtmf_result = await GetDtmfTask(
num_digits=4,
chat_ctx=self.chat_ctx,
)WarmTransferTask
Executes an agent-assisted warm transfer: dialing a human supervisor, providing conversation context, and merging calls. This handles creating a separate room for the human agent, managing hold music, and coordinating the handoff:
from livekit.agents.beta.workflows import WarmTransferTask
transfer_result = await WarmTransferTask(
transfer_to="+15551234567",
chat_ctx=self.chat_ctx,
)Customizing prebuilt tasks
All prebuilt tasks accept extra_instructions and tools parameters for customization without subclassing:
email_result = await GetEmailTask(
chat_ctx=self.chat_ctx,
extra_instructions="If the user cannot provide an email, ask for a phone number instead.",
tools=[get_alternative_contact],
)Testing Tasks
Tasks are self-contained units with clear inputs and outputs, which makes them straightforward to test. The key challenge is mocking the RunContext that tool methods receive.
Mocking RunContext
The RunContext object carries speech_handle and session references. For unit tests, use MagicMock:
import pytest
from unittest.mock import MagicMock, AsyncMock
from my_agent.tasks import CollectPhoneNumberTask
@pytest.fixture
def mock_context():
ctx = MagicMock(spec=RunContext)
ctx.speech_handle = MagicMock() # Unique handle per "turn"
return ctx
@pytest.fixture
def task():
task = CollectPhoneNumberTask()
task.session = MagicMock()
task.session.generate_reply = AsyncMock()
return taskTesting the two-step flow
The two-step confirmation is the most important thing to test. Verify that the task rejects same-turn confirmation and accepts cross-turn confirmation:
@pytest.mark.asyncio
async def test_confirm_rejects_same_turn(task, mock_context):
"""confirm_phone_number should reject if called in the same turn as record."""
# Record the phone number
result = await task.record_phone_number(
mock_context, phone_number="5551234567", country_code="+1"
)
assert "Read this back" in result
# Try to confirm with the same speech_handle (same LLM turn)
with pytest.raises(ToolError, match="Wait for the caller"):
await task.confirm_phone_number(mock_context)
@pytest.mark.asyncio
async def test_confirm_accepts_different_turn(task, mock_context):
"""confirm_phone_number should succeed when called in a new turn."""
# Record the phone number
await task.record_phone_number(
mock_context, phone_number="5551234567", country_code="+1"
)
# Simulate a new turn by changing the speech_handle
new_ctx = MagicMock(spec=RunContext)
new_ctx.speech_handle = MagicMock() # Different handle = new turn
await task.confirm_phone_number(new_ctx)
assert task.done()Testing ToolError cases
Validate that your input validation produces clear, actionable error messages:
@pytest.mark.asyncio
async def test_rejects_short_phone_number(task, mock_context):
"""Should reject phone numbers that are too short."""
with pytest.raises(ToolError, match="too short"):
await task.record_phone_number(
mock_context, phone_number="123", country_code="+1"
)
@pytest.mark.asyncio
async def test_rejects_confirm_without_record(task, mock_context):
"""Should reject confirmation if no phone number was recorded."""
with pytest.raises(ToolError, match="No phone number recorded"):
await task.confirm_phone_number(mock_context)Testing optional skip paths
@pytest.mark.asyncio
async def test_skip_hours_completes_with_required_fields(task, mock_context):
"""Task should complete when required fields are set and hours are skipped."""
await task.record_services(mock_context, services=["plumbing", "drain cleaning"])
await task.record_location(mock_context, location="North Austin")
await task.skip_hours(mock_context)
assert task.done()
result = task.result()
assert result.services == ["plumbing", "drain cleaning"]
assert result.location == "North Austin"
assert result.hours is NoneTesting the readback gate
@pytest.mark.asyncio
async def test_confirm_blocked_without_readback(task, mock_context):
"""confirm_details should fail if readback_complete was not called."""
with pytest.raises(ToolError, match="read back all details first"):
await task.confirm_details(mock_context, confirmed=True)
@pytest.mark.asyncio
async def test_update_resets_readback(task, mock_context):
"""Updating a field should require a fresh readback."""
# Complete readback
await task.readback_complete(mock_context)
# Update a field
new_ctx = MagicMock(spec=RunContext)
new_ctx.speech_handle = MagicMock()
await task.update_field(new_ctx, field_name="business_name", new_value="New Name")
# Readback flag should be reset
with pytest.raises(ToolError, match="read back all details first"):
another_ctx = MagicMock(spec=RunContext)
another_ctx.speech_handle = MagicMock()
await task.confirm_details(another_ctx, confirmed=True)Putting It All Together
Here is how these pieces compose in a production agent. The agent has a single tool that kicks off the entire multi-step collection flow:
from livekit.agents import Agent, function_tool
from livekit.agents.beta.workflows import TaskGroup
from livekit.agents import RunContext
class OnboardingAgent(Agent):
def __init__(self) -> None:
super().__init__(
instructions="You are Lisa, a friendly onboarding assistant for tradespeople."
)
async def on_enter(self) -> None:
self.session.generate_reply(
instructions="Greet the caller and offer to help set up their business profile."
)
@function_tool()
async def collect_business_profile(self, ctx: RunContext) -> str:
"""Collect the caller's full business profile."""
# Phase 1: Sequential collection via TaskGroup
collect_group = TaskGroup(chat_ctx=self.chat_ctx)
collect_group.add(
lambda: CollectBusinessIdentityTask(),
id="identity",
description="Business name and trade type",
)
collect_group.add(
lambda: CollectOperationsTask(),
id="operations",
description="Services, hours, and location",
)
collect_group.add(
lambda: CollectOwnerContactTask(),
id="contact",
description="Owner name and contact info",
)
results = await collect_group
r = results.task_results
# Assemble the full profile
profile = BusinessProfileResult(
business_name=r["identity"].business_name,
trade_type=r["identity"].trade_type,
services=r["operations"].services,
hours=r["operations"].hours or "",
location=r["operations"].location,
owner_name=r["contact"].owner_name,
owner_phone=r["contact"].owner_phone,
email_address=r["contact"].email_address,
)
# Phase 2: Readback and confirmation
confirmed_profile = await ConfirmDetailsTask(profile=profile)
return f"Profile confirmed for {confirmed_profile.business_name}."The flow is clean: three collection tasks run in sequence inside a TaskGroup, then a standalone confirmation task handles the readback gate. Each task manages its own validation, its own two-step confirmation where needed, and its own skip paths for optional fields. The orchestrating agent does not need to know any of those details. It just awaits the results.
This is the power of the task system. Complex, multi-step voice workflows decompose into focused, testable, reusable units. The two-step confirmation pattern guarantees the caller actually confirms. The readback gate guarantees the agent reads back details before confirming. And TaskGroup handles the sequencing, context sharing, and regression support.
In the next chapter, we will look at multi-agent workflows: what happens when a single agent is not enough and you need to hand off between specialized agents during a call.