LiveKit docs › Logic & Structure › Tasks & task groups

---

# Tasks and task groups

> Use tasks to build complex workflows for your voice AI agents.

## Overview

Tasks are focused, reusable units that perform a specific objective and return a typed result. They run inside an agent and take control of the session only until their goal is achieved. A task can define its own [tools](https://docs.livekit.io/agents/logic/tools.md) and starts executing when it's created within the context of an agent.

For multi-step flows, the framework provides `TaskGroup`. A task group executes an ordered sequence of tasks while allowing users to return to earlier steps for corrections. All tasks in a group share conversation context, and when the group finishes, a summarized result is returned to the agent that started it.

Tasks and task groups are core building blocks for complex voice AI [workflows](https://docs.livekit.io/agents/logic/workflows.md). Reach for them whenever you want a guided, structured conversation that returns a typed result, for example:

- Qualifying a lead.
- Collecting patient intake information.
- Running a follow-up survey or feedback call.
- Gathering booking or service-request details.
- Collecting structured information such as an address or payment details.
- Obtaining recording consent at the start of a call.
- Walking through a series of questions one step at a time.
- Any discrete action that should complete and yield control.

You can build a structured collection flow two ways:

- **With the SDK**: compose `AgentTask`s and `TaskGroup`s using the LiveKit Agents SDK in Python or Node.js. Use this when you want fine-grained control, want to reuse the same collection step across multiple agents, or are composing structured collection into a larger code-first agent.
- **In Agent Builder**: configure fields in [Data Collection mode](https://docs.livekit.io/agents/start/builder.md#data-collection) to prototype a flow in the browser. Builder compiles the configuration into the same `AgentTask` and `TaskGroup` primitives documented on this page, so you can [download the code](https://docs.livekit.io/agents/start/builder.md#convert-to-code) when you need to extend it.

> 💡 **Prebuilt tasks**
> 
> See [Prebuilt tasks](https://docs.livekit.io/agents/prebuilt/tasks.md) for ready-to-use task components such as email collection, address capture, DTMF input, and warm transfer. Use these alongside your own custom tasks inside a `TaskGroup`.

## Defining a task

Define a task by extending the `AgentTask` class and specifying a result type using [generics](https://typing.python.org/en/latest/reference/generics.html) (Python) or TypeScript [generics](https://www.typescriptlang.org/docs/handbook/2/generics.html) (Node.js). Use the `on_enter` method to begin the task's interaction with the user, and call the `complete` method with a result when finished. The task has full support for tools, similar to an agent.

**Python**:

```python
from livekit.agents import AgentTask, function_tool

class CollectConsent(AgentTask[bool]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="""
            Ask for recording consent and get a clear yes or no answer.
            Be polite and professional.
            """,
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="""
            Briefly introduce yourself, then ask for permission to record the call for quality assurance and training purposes.
            Make it clear that they can decline.
            """
        )

    @function_tool
    async def consent_given(self) -> None:
        """Use this when the user gives consent to record."""
        self.complete(True)

    @function_tool
    async def consent_denied(self) -> None:
        """Use this when the user denies consent to record."""
        self.complete(False)

```

---

**Node.js**:

```tsx
import { llm, voice } from '@livekit/agents';

class CollectConsent extends voice.AgentTask<boolean> {
  constructor(chatCtx?: llm.ChatContext) {
    super({
      instructions: `
        Ask for recording consent and get a clear yes or no answer.
        Be polite and professional.
      `,
      chatCtx,
      tools: {
        consentGiven: llm.tool({
          description: 'Use this when the user gives consent to record.',
          execute: async () => {
            this.complete(true);
          },
        }),
        consentDenied: llm.tool({
          description: 'Use this when the user denies consent to record.',
          execute: async () => {
            this.complete(false);
          },
        }),
      },
    });
  }

  async onEnter(): Promise<void> {
    await this.session.generateReply({
      instructions: `
        Briefly introduce yourself, then ask for permission to record
        the call for quality assurance and training purposes.
        Make it clear that they can decline.
      `,
    });
  }
}

```

### Running a task

A task must be created within the context of an [active](https://docs.livekit.io/agents/logic/agents-handoffs.md#active-agent) `Agent`, and runs automatically when it's created. The task takes control of the session until it returns a result. Await the task to receive its result.

A task can only be awaited from one of three call sites in agent code:

- **`on_enter`**: runs the task as the agent becomes active. Useful for deterministic setup steps.
- **`on_exit`**: runs the task as the agent becomes inactive. Useful for wrap-up steps before a handoff or session end.
- **A tool function body**: the tool instantiates and awaits the task. The LLM decides when to invoke the tool, so delegation happens mid-conversation.

Awaiting an `AgentTask` outside these call sites raises a `RuntimeError`.

**Python**:

```python
from livekit import api
from livekit.agents import Agent, function_tool, get_job_context

class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a friendly customer service representative.")

    async def on_enter(self) -> None:
        if await CollectConsent(chat_ctx=self.chat_ctx):
            await self.session.generate_reply(instructions="Offer your assistance to the user.")
        else:
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            job_ctx = get_job_context()
            await job_ctx.api.room.delete_room(api.DeleteRoomRequest(room=job_ctx.room.name))

```

---

**Node.js**:

```tsx
import { voice } from '@livekit/agents';

class CustomerServiceAgent extends voice.Agent {
  constructor() {
    super({ instructions: 'You are a friendly customer service representative.' });
  }

  async onEnter(): Promise<void> {
    const consent = await new CollectConsent(this.chatCtx).run();

    if (consent) {
      await this.session.generateReply({
        instructions: 'Offer your assistance to the user.',
      });
    } else {
      await this.session.generateReply({
        instructions: 'Inform the user that you are unable to proceed and will end the call.',
      });
      this.session.shutdown({ reason: 'user-ended-call' });
    }
  }
}

```

> ℹ️ **Testing limitation**
> 
> `get_job_context()` is unavailable in test environments and raises a `RuntimeError` when called. If your agent uses `get_job_context()`, avoid testing code paths that invoke it, or mock the call using `unittest.mock`.

### Passing conversation history to a task

By default, a task starts with an empty chat context. To include the parent agent's conversation history, pass `chat_ctx` to the task constructor. Use `exclude_instructions=True` (Python) or `excludeInstructions: true` (Node.js) to omit the parent's system prompt so the task's own instructions take effect:

**Python**:

```python
class GetContactInfoTask(AgentTask[ContactInfoResult]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="Collect the user's name, email address, and phone number.",
            chat_ctx=chat_ctx,
        )
    # ....

class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a friendly customer service representative.")

    @function_tool()
    async def collect_contact_info(self):
        """Collect the user's contact information."""
        # Pass conversation history without the agent's system prompt
        result = await GetContactInfoTask(
            chat_ctx=self.chat_ctx.copy(exclude_instructions=True)
        )
        return f"Recorded contact info for {result.name}."

```

---

**Node.js**:

```tsx
class GetContactInfoTask extends voice.AgentTask<ContactInfoResult> {
  constructor(chatCtx?: llm.ChatContext) {
    super({
      instructions: "Collect the user's name, email address, and phone number.",
      chatCtx,
    });
  }
  // ....
}

class CustomerServiceAgent extends voice.Agent {
  constructor() {
    super({
      instructions: 'You are a friendly customer service representative.',
      tools: {
        collectContactInfo: llm.tool({
          description: "Collect the user's contact information.",
          execute: async (_, { ctx }) => {
            // Pass conversation history without the agent's system prompt
            const result = await new GetContactInfoTask(
              ctx.session.chatCtx.copy({ excludeInstructions: true }),
            ).run();
            return `Recorded contact info for ${result.name}.`;
          },
        }),
      },
    });
  }
}

```

The `copy()` method also accepts additional filters like `exclude_function_call` and `exclude_handoff`. For a complete list of available filters, refer to [Copying a context](https://docs.livekit.io/agents/logic/chat-context.md#copying-a-context).

### Task results

Use any result type you want. For complex results, use a custom dataclass (Python) or interface (Node.js).

**Python**:

```python
from dataclasses import dataclass

@dataclass
class ContactInfoResult:
    name: str
    email_address: str
    phone_number: str

class GetContactInfoTask(AgentTask[ContactInfoResult]):
    # ....

```

---

**Node.js**:

```tsx
interface ContactInfoResult {
  name: string;
  emailAddress: string;
  phoneNumber: string;
}

class GetContactInfoTask extends voice.AgentTask<ContactInfoResult> {
  // ....
}

```

### Unordered collection within tasks

You can use a single task to collect multiple pieces of information in any order. The following example collects strengths, weaknesses, and work style in a hypothetical interview. Candidates can answer the questions in any order:

**Python**:

```python
@dataclass
class BehavioralResults:
    strengths: str
    weaknesses: str
    work_style: str

class BehavioralTask(AgentTask[BehavioralResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Collect strengths, weaknesses, and work style in any order."
        )
        self._results = {}
    
    @function_tool()
    async def record_strengths(self, strengths_summary: str):
        """Record candidate's strengths"""
        self._results["strengths"] = strengths_summary
        self._check_completion()
    
    @function_tool()
    async def record_weaknesses(self, weaknesses_summary: str):
        """Record candidate's weaknesses"""
        self._results["weaknesses"] = weaknesses_summary
        self._check_completion()
    
    @function_tool()
    async def record_work_style(self, work_style: str):
        """Record candidate's work style"""
        self._results["work_style"] = work_style
        self._check_completion()
    
    def _check_completion(self):
        required_keys = {"strengths", "weaknesses", "work_style"}
        if self._results.keys() == required_keys:
            results = BehavioralResults(
                strengths=self._results["strengths"],
                weaknesses=self._results["weaknesses"],
                work_style=self._results["work_style"]
            )
            self.complete(results)
        else:
            self.session.generate_reply(
                instructions="Continue collecting remaining information."
            )

```

---

**Node.js**:

```tsx
import { llm, voice } from '@livekit/agents';
import { z } from 'zod';

interface BehavioralResults {
  strengths: string;
  weaknesses: string;
  workStyle: string;
}

class BehavioralTask extends voice.AgentTask<BehavioralResults> {
  private results: Partial<BehavioralResults> = {};

  constructor() {
    super({
      instructions: 'Collect strengths, weaknesses, and work style in any order.',
      tools: {
        recordStrengths: llm.tool({
          description: "Record candidate's strengths",
          parameters: z.object({
            strengthsSummary: z.string().describe("Summary of candidate's strengths"),
          }),
          execute: async ({ strengthsSummary }) => {
            this.results.strengths = strengthsSummary;
            this.checkCompletion();
          },
        }),
        recordWeaknesses: llm.tool({
          description: "Record candidate's weaknesses",
          parameters: z.object({
            weaknessesSummary: z.string().describe("Summary of candidate's weaknesses"),
          }),
          execute: async ({ weaknessesSummary }) => {
            this.results.weaknesses = weaknessesSummary;
            this.checkCompletion();
          },
        }),
        recordWorkStyle: llm.tool({
          description: "Record candidate's work style",
          parameters: z.object({
            workStyle: z.string().describe("Description of candidate's work style"),
          }),
          execute: async ({ workStyle }) => {
            this.results.workStyle = workStyle;
            this.checkCompletion();
          },
        }),
      },
    });
  }

  private checkCompletion(): void {
    const { strengths, weaknesses, workStyle } = this.results;
    if (strengths && weaknesses && workStyle) {
      this.complete({ strengths, weaknesses, workStyle });
    } else {
      this.session.generateReply({
        instructions: 'Continue collecting remaining information.',
      });
    }
  }
}

```

## Task group

> 🔥 **Experimental feature**
> 
> `TaskGroup` is currently experimental and the API might change in a future release.

Task groups let you build complex, user-friendly workflows that mirror real conversational behavior — where users might need to revisit or correct earlier steps without losing context. They're designed as ordered, multi-step flows that can be broken into discrete tasks, with built-in regression support for safely moving backward.

`TaskGroup` supports task chaining, which allows tasks to call or re-enter other tasks dynamically while maintaining the overall flow order. This lets users return to earlier steps as often as needed. All tasks in the group share the same conversation context, and when the group finishes, the summarized context can be passed back to the controlling agent.

### Configuration options

`TaskGroup` supports the following parameters:

- **`summarize_chat_ctx`** _(boolean)_ (optional) - Default: `true`: Whether to summarize the interactions within the `TaskGroup` into one message and merge into the main context.

- **`chat_ctx`** _(llm.ChatContext)_ (optional) - Default: `llm.ChatContext`: The shared [chat context](https://docs.livekit.io/agents/logic/chat-context.md) within the TaskGroup. Pass the current chat context to ensure conversational continuity.

- **`return_exceptions`** _(boolean)_ (optional) - Default: `false`: Controls error handling when a sub-task raises an unhandled exception. When set to `true`, the exception is added to the results dictionary and the sequence continues. When set to `false`, the exception propagates immediately and the sequence stops.

- **`on_task_completed`** _((event: TaskCompletedEvent) => Promise<void>)_ (optional): An async callback invoked after each sub-task completes successfully. It receives a `TaskCompletedEvent` with the following fields:

- `agent_task`: `AgentTask` instance that just finished.
- `task_id`: String ID of the task.
- `result`: Value the task returned.

### Basic usage

Initialize and set up a `TaskGroup` by adding tasks to it. Add tasks in the order they should be executed:

**Python**:

```python
from livekit.agents.beta.workflows import GetEmailTask, TaskGroup


# Create and configure TaskGroup with the current agent's chat context
chat_ctx = self.chat_ctx
task_group = TaskGroup(chat_ctx=chat_ctx)

# Add tasks using lambda factories
task_group.add(
    lambda: GetEmailTask(), 
    id="get_email_task", 
    description="Collects the user's email"
)
task_group.add(
    lambda: GetCommuteTask(), 
    id="get_commute_task", 
    description="Records the user's commute flexibility"
)

# Execute the task group
results = await task_group  # Returns TaskGroupResult object
task_results = results.task_results

# Access results by task ID
print(task_results)
# Output: {
#   "get_email_task": GetEmailResult(email="john.doe@gmail.com"), 
#   "get_commute_task": CommuteResult(can_commute=True, commute_method="subway")
# }

```

---

**Node.js**:

```tsx
import { beta, llm } from '@livekit/agents';

// Create and configure TaskGroup with the current agent's chat context
const chatCtx = this.chatCtx;
const taskGroup = new beta.TaskGroup({ chatCtx });

// Add tasks using arrow-function factories
taskGroup.add(() => new GetEmailTask(), {
  id: 'get_email_task',
  description: "Collects the user's email",
});
taskGroup.add(() => new GetCommuteTask(), {
  id: 'get_commute_task',
  description: "Records the user's commute flexibility",
});

// Execute the task group
const results = await taskGroup.run(); // Returns TaskGroupResult object
const taskResults = results.taskResults;

// Access results by task ID
console.log(taskResults);
// Output: {
//   get_email_task: { email: "john.doe@gmail.com" },
//   get_commute_task: { canCommute: true, commuteMethod: "subway" }
// }

```

The `TaskGroup.add()` method takes a task factory and an options object (Python: `task_factory`, `id`, `description` as arguments; Node.js: factory function and `{ id, description }`):

- **Task factory**: A callable that returns a task instance (Python: typically a lambda; Node.js: an arrow function).
- **id**: A string identifier for the task used to access results.
- **description**: A string description that helps the LLM understand when to regress to this task.

The factory allows for tasks to be reinitialized with the same arguments when revisited. The task id and description are passed to the LLM as task identifiers when the LLM needs to regress to a previous task. This allows the LLM to understand the task's purpose and context when revisiting it. Task chaining is supported, allowing users to return to earlier steps as often as needed.

All tasks share the same conversation context. The context is summarized and passed back to the controlling agent when the group finishes. This option can be disabled when initializing the task group:

**Python**:

```python
# Disable context summarization
task_group = TaskGroup(summarize_chat_ctx=False)

```

---

**Node.js**:

```tsx
// Disable context summarization
const taskGroup = new beta.TaskGroup({ summarizeChatCtx: false });

```

### Task completion callbacks

Add a callback function to a task group to run custom logic after each task completes. The callback receives a `TaskCompletedEvent` containing the completed task's ID, instance, and result.

Use the `on_task_completed` parameter to set the callback function. The following example prints a message after each task finishes:

**Python**:

```python
from livekit.agents.beta.workflows import TaskGroup, TaskCompletedEvent

async def print_task_result(event: TaskCompletedEvent) -> None:
    print(f"Task '{event.task_id}' completed with result: {event.result}")

task_group = TaskGroup(
    chat_ctx=self.chat_ctx,
    on_task_completed=print_task_result,
)
task_group.add(
    lambda: IntroTask(),
    id="intro_task",
    description="Collects name and introduction",
)
task_group.add(
    lambda: CommuteTask(),
    id="commute_task",
    description="Asks about commute flexibility",
)

results = await task_group

```

---

**Node.js**:

```tsx
import { beta } from '@livekit/agents';

const taskGroup = new beta.TaskGroup({
  chatCtx: this.chatCtx,
  onTaskCompleted: async ({ taskId, result }) => {
    console.log(`Task '${taskId}' completed with result:`, result);
  },
});
taskGroup.add(() => new IntroTask(), {
  id: 'intro_task',
  description: 'Collects name and introduction',
});
taskGroup.add(() => new CommuteTask(), {
  id: 'commute_task',
  description: 'Asks about commute flexibility',
});

const results = await taskGroup.run();

```

### Early exit from a task group

Avoid calling `session.shutdown()` directly from `on_task_completed`. The callback runs while `TaskGroup` is still iterating its task stack. Because the group hasn't finished yet, shutting down the session at that point raises a `RuntimeError`.

To skip the remaining tasks when an earlier task signals an exit condition, raise a custom exception from the callback and catch it where you await the task group. With the default `return_exceptions=False`, `TaskGroup` propagates the exception to the awaiting code:

**Python**:

```python
from livekit.agents.beta.workflows import TaskGroup, TaskCompletedEvent

class ExistingProfileFound(Exception):
    """Raised to skip remaining intake tasks when a returning user is detected."""

async def check_for_existing_profile(event: TaskCompletedEvent) -> None:
    if event.task_id == "get_email_task":
        if database.find_user_by_email(event.result.email_address):
            raise ExistingProfileFound()

task_group = TaskGroup(
    chat_ctx=self.chat_ctx,
    on_task_completed=check_for_existing_profile,
)
task_group.add(lambda: GetEmailTask(), id="get_email_task", description="Collects the user's email")
task_group.add(lambda: GetAddressTask(), id="get_address_task", description="Collects the user's address")

try:
    results = await task_group
except ExistingProfileFound:
    # Safe to run cleanup logic here — the task group is no longer iterating
    await self.session.generate_reply(instructions="Welcome the returning user.")

```

---

**Node.js**:

Prebuilt tasks aren't available in Node.js. In the following example, you must [define your own tasks](#define-task) for `GetEmailTask` and `GetAddressTask`:

```tsx
import { beta } from '@livekit/agents';

class ExistingProfileFound extends Error {}

const taskGroup = new beta.TaskGroup({
  chatCtx: this.chatCtx,
  onTaskCompleted: async ({ taskId, result }) => {
    if (taskId === 'get_email_task') {
      const { email } = result as { email: string };
      if (database.findUserByEmail(email)) {
        throw new ExistingProfileFound();
      }
    }
  },
});

taskGroup.add(() => new GetEmailTask(), {
  id: 'get_email_task',
  description: "Collects the user's email",
});
taskGroup.add(() => new GetAddressTask(), {
  id: 'get_address_task',
  description: "Collects the user's address",
});

try {
  const results = await taskGroup.run();
} catch (e) {
  if (e instanceof ExistingProfileFound) {
    // Safe to run cleanup logic here — the task group is no longer iterating
    await this.session.generateReply({ instructions: 'Welcome the returning user.' });
  } else {
    throw e;
  }
}

```

This pattern requires the default value `return_exceptions=False`. When `return_exceptions` is `True`, `TaskGroup` stores the exception in the results dictionary and continues the sequence instead of stopping.

`TaskGroup` uses the same exception-based mechanism internally to handle [regression](#taskgroup) — when the LLM requests to revisit an earlier task, the active task is completed with an internal exception that the group catches and uses to reorder the task stack.

### Complete workflow example

The following is a complete example showing how to build an interview workflow with `TaskGroup`. It collects basic candidate information and then asks about their commute flexibility:

**Python**:

```python
from livekit.agents import AgentTask, function_tool, RunContext
from livekit.agents.beta.workflows import TaskGroup
from dataclasses import dataclass

@dataclass
class IntroResults:
    name: str
    intro: str

@dataclass 
class CommuteResults:
    can_commute: bool
    commute_method: str

class IntroTask(AgentTask[IntroResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Welcome the candidate and collect their name and introduction."
        )
    
    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="Welcome the candidate and gather their name."
        )
    
    @function_tool()
    async def record_intro(self, context: RunContext, name: str, intro_notes: str) -> None:
        """Record the candidate's name and introduction"""
        context.session.userdata.candidate_name = name
        results = IntroResults(name=name, intro=intro_notes)
        self.complete(results)

class CommuteTask(AgentTask[CommuteResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Ask about the candidate's ability to commute to the office."
        )
    
    @function_tool()
    async def record_commute_flexibility(
        self, 
        context: RunContext, 
        can_commute: bool, 
        commute_method: str
    ) -> None:
        """Record commute flexibility and transportation method"""
        results = CommuteResults(can_commute=can_commute, commute_method=commute_method)
        self.complete(results)

# Set up the workflow
task_group = TaskGroup()
task_group.add(
    lambda: IntroTask(), 
    id="intro_task", 
    description="Collects name and introduction"
)
task_group.add(
    lambda: CommuteTask(), 
    id="commute_task", 
    description="Asks about commute flexibility"
)

# Execute and get results
results = await task_group
task_results = results.task_results

```

---

**Node.js**:

```tsx
import { beta, llm, voice } from '@livekit/agents';
import { z } from 'zod';

interface IntroResults {
  name: string;
  intro: string;
}

interface CommuteResults {
  canCommute: boolean;
  commuteMethod: string;
}

interface InterviewUserData {
  candidateName?: string;
}

class IntroTask extends voice.AgentTask<IntroResults, InterviewUserData> {
  constructor() {
    super({
      instructions: 'Welcome the candidate and collect their name and introduction.',
      tools: {
        recordIntro: llm.tool({
          description: "Record the candidate's name and introduction",
          parameters: z.object({
            name: z.string().describe("The candidate's name"),
            introNotes: z.string().describe('Introduction notes'),
          }),
          execute: async ({ name, introNotes }, { ctx }) => {
            ctx.userData.candidateName = name;
            this.complete({ name, intro: introNotes });
          },
        }),
      },
    });
  }

  async onEnter(): Promise<void> {
    await this.session.generateReply({
      instructions: 'Welcome the candidate and gather their name.',
    });
  }
}

class CommuteTask extends voice.AgentTask<CommuteResults> {
  constructor() {
    super({
      instructions: "Ask about the candidate's ability to commute to the office.",
      tools: {
        recordCommuteFlexibility: llm.tool({
          description: 'Record commute flexibility and transportation method',
          parameters: z.object({
            canCommute: z.boolean().describe('Whether the candidate can commute'),
            commuteMethod: z.string().describe('Transportation method'),
          }),
          execute: async ({ canCommute, commuteMethod }) => {
            this.complete({ canCommute, commuteMethod });
          },
        }),
      },
    });
  }
}

// Set up the workflow
const taskGroup = new beta.TaskGroup();
taskGroup.add(() => new IntroTask(), {
  id: 'intro_task',
  description: 'Collects name and introduction',
});
taskGroup.add(() => new CommuteTask(), {
  id: 'commute_task',
  description: 'Asks about commute flexibility',
});

// Execute and get results
const results = await taskGroup.run();
const taskResults = results.taskResults;

```

### Best practices for testing task groups

The following sections provide specific guidelines for testing `TaskGroup` in both Python and Node.js SDKs.

#### Add a short delay before the first session.run() in Python

`TaskGroup` temporarily sets `llm=None` during task transitions. In the Python SDK, `session.run()` doesn't fall back to `session.llm` during this window, which can raise the following exception if the test calls `session.run()` too early:

`RuntimeError: trying to generate reply without an LLM model.`

Add a small delay between `session.start()` and the first `session.run()` call so the first sub-task can take over:

```python
await session.start()
await asyncio.sleep(0.5)
await session.run(...)

```

This delay isn't required in Node.js because `null` LLM values automatically fall back to `session.llm`.

#### Parse function call arguments

Test run results store function call arguments as raw JSON strings. The built-in assertion helpers (`is_function_call`, `contains_function_call` in Python, and `isFunctionCall`, `containsFunctionCall` in Node.js) parse the JSON for you and support partial-dict matching, so prefer them when checking known argument values.

Parse the JSON manually only when you need an assertion the helpers can't express, such as, range checks, regular expression matches, or comparisons against a value computed in the test.

**Python**:

Use the helper for direct value matches:

```python
result.expect.contains_function_call(
    name="record_commute",
    arguments={"can_commute": True},
)

```

Parse `item.arguments` only when you need a richer assertion:

```python
fnc = result.expect.contains_function_call(name="record_experience")
args = json.loads(fnc.event().item.arguments)
assert args["years_of_experience"] >= 5

```

---

**Node.js**:

Use the helper for direct value matches:

```typescript
result.expect.containsFunctionCall({
  name: 'recordCommute',
  args: { canCommute: true },
});

```

Parse `item.args` only when you need a richer assertion:

```typescript
const fnc = result.expect.containsFunctionCall({ name: 'recordExperience' });
const args = JSON.parse(fnc.event().item.args);
expect(args.yearsOfExperience).toBeGreaterThanOrEqual(5);

```

#### Initialize userData when tasks depend on it

If tasks read or write `ctx.userData`, initialize it when creating the session. The failure mode differs across SDKs:

- **Python:** accessing `session.userdata` when it's unset raises `ValueError: AgentSession userdata is not set`.
- **Node.js:** accessing `session.userData` when it's unset throws `Error: Voice agent userData is not set`.

In either case, pass an initialized value to the session constructor:

**Python**:

```python
AgentSession(llm=llm, userdata=MyUserdata(candidate_name=""))

```

---

**Node.js**:

```typescript
new voice.AgentSession<MyUserData>({ llm, userData: { candidateName: '' } });

```

#### Don't assert on startup output

Output generated during agent startup (for example from `session.say()` or `session.generate_reply()`) is not included in `RunResult`.

Structure tests to assert agent responses to user input, not startup messages.

#### Avoid awaiting playout inside onEnter() when triggered from a tool

If `onEnter()` runs inside a tool's `execute` function, awaiting speech playout can cause a circular wait. The tool call remains active until `onEnter()` returns.

Call `generateReply()` without awaiting it:

**Python**:

```python
async def on_enter(self) -> None:
    self.session.generate_reply(instructions="Welcome the user.")  # don't await

```

---

**Node.js**:

```typescript
async onEnter(): Promise<void> {
  this.session.generateReply({ instructions: 'Welcome the user.' }); // no await
}

```

#### Consider multi-turn LLM behavior

An LLM might not call a task's completion tool on the first turn. It might require multiple exchanges before completing the task.

Prefer `containsFunctionCall()` over `nextEvent()` for more resilient tests, and use generous timeouts:

- `containsFunctionCall()` checks whether the call occurred anywhere in the response.
- `nextEvent()` only checks the immediate next event.

**Python**:

```python
result.expect.contains_function_call(name="consent_given")

```

---

**Node.js**:

```typescript
result.expect.containsFunctionCall({ name: 'consentGiven' });

```

#### Increase cleanup timeouts in Node.js

Session cleanup can be slow when a `TaskGroup` is mid-flow. Set an explicit timeout in your cleanup hook to avoid `afterEach` failures:

```typescript
afterEach(async () => {
  await session?.close();
}, 30000);

```

#### Example tests for task group

- **[Testing a task group (Node.js)](https://github.com/livekit/agents-js/blob/main/examples/src/testing/basic_task_group.test.ts)**: This test suite verifies the behavior of the basic task group example in the Node.js GitHub repo.

## Examples

The following examples show tasks and task groups in production-style agents:

- **[Survey agent (Python)](https://github.com/livekit/agents/blob/main/examples/survey/agent.py)**: Interview screening agent that runs a TaskGroup of five tasks: intro, email capture, commute, experience, and behavioral. Uses session userdata, a disqualify tool, CSV export, and post-interview LLM evaluation.

- **[Basic agent task (Node.js)](https://github.com/livekit/agents-js/blob/main/examples/src/basic_agent_task.ts)**: Survey agent that runs reusable `AgentTask`s from `onEnter` and from tools. Uses a generic info-collection task, then shows handoff to a separate weather agent and back.

- **[Basic task group (Node.js)](https://github.com/livekit/agents-js/blob/main/examples/src/basic_task_group.ts)**: Onboarding agent that starts a two-step TaskGroup (name then email) via a tool. Demonstrates `onTaskCompleted`, context summarization, and regression so users can correct earlier answers (e.g. "change my name to …").

## Additional resources

The following topics provide more information on creating complex workflows for your voice AI agents.

- **[Workflows](https://docs.livekit.io/agents/logic/workflows.md)**: Complete guide to defining and using workflows in your agents.

- **[Data Collection mode in Agent Builder](https://docs.livekit.io/agents/start/builder.md#data-collection)**: Build a structured data collection agent right from your browser.

- **[Tool definition and use](https://docs.livekit.io/agents/logic/tools.md)**: Complete guide to defining and using tools in your agents.

- **[Nodes](https://docs.livekit.io/agents/logic/nodes.md)**: Add custom behavior to any component of the voice pipeline.

- **[Testing & evaluation](https://docs.livekit.io/agents/start/testing.md)**: Test every aspect of your agents with a custom test suite.

---

This document was rendered at 2026-06-07T11:33:39.230Z.
For the latest version of this document, see [https://docs.livekit.io/agents/logic/tasks.md](https://docs.livekit.io/agents/logic/tasks.md).

To explore all LiveKit documentation, see [llms.txt](https://docs.livekit.io/llms.txt).