diff --git a/examples/multi-worker/ui-worker/async-tasks/README.md b/examples/multi-worker/ui-worker/async-tasks/README.md
new file mode 100644
index 000000000..2ace60701
--- /dev/null
+++ b/examples/multi-worker/ui-worker/async-tasks/README.md
@@ -0,0 +1,90 @@
+# async-tasks
+
+The UIWorker fans out long-running work to multiple peer workers in
+parallel, streams their progress to an in-flight panel on the page, and
+lets the user cancel mid-flight.
+
+## What it shows
+
+- The **`user_job_group` / `start_user_job_group`** API: dispatching
+ parallel work to multiple peer workers and automatically forwarding
+ every job lifecycle event to the client. The `reply` tool calls
+ `start_user_job_group("wikipedia", "news", "scholar", payload=...,
+ label=...)` and the `UIWorker` does the rest.
+- The four **`ui-task` envelopes** the worker forwards (`group_started`,
+ `task_update`, `task_completed`, `group_completed`) and the
+ client-side `RTVIEvent.UITask` event for consuming them. The client
+ keeps a state map keyed by `task_id` and renders per-worker progress.
+- **Cancellation**: the in-flight card's Cancel button calls
+ `client.cancelUITask(task_id, reason)`. The reserved `__cancel_task`
+ event is translated by the `UIWorker` into `cancel_job_group(task_id)`
+ on the registered group; cancelled workers report status `cancelled`.
+- **Background dispatch from a tool**: `start_user_job_group` returns
+ immediately so the `reply` tool can speak its acknowledgement
+ ("Researching the Mariana Trench now") while the workers run — the
+ main LLM is free to take follow-up turns.
+
+## What it adds vs. the prior demos
+
+The other examples use the request/response half of the bus protocol
+(main LLM → UIWorker → reply). This one adds the streaming job-group
+half: UIWorker → peer workers → progress events forwarded to the client.
+The architecture grows from "one delegate" to "one delegate plus a
+worker pool" — the peers are plain `BaseWorker`s launched on the runner.
+
+## Run
+
+Two terminals.
+
+**Terminal 1 — bot:**
+
+```bash
+cd examples/multi-worker/ui-worker/async-tasks
+uv run python bot.py
+```
+
+The bot starts on `http://localhost:7860`.
+
+**Terminal 2 — client:**
+
+```bash
+cd examples/multi-worker/ui-worker/async-tasks/client
+npm install # one-time
+npm run dev
+```
+
+Open `http://localhost:5173` and click **Connect**.
+
+## What to try
+
+The workers are simulated (canned summaries, randomized `asyncio.sleep`
+delays) so the demo focuses on the protocol, not the AI. Each research
+call takes a few seconds.
+
+- _"Research the Mariana Trench."_ — the worker spawns three peers,
+ acknowledges in one short reply, and a card appears showing each
+ peer's status as it progresses (searching → found N results →
+ summarizing → completed).
+- _"Look up octopus cognition."_ — same flow; a second card stacks.
+- _"Research the moon, then research Mars."_ — two groups run
+ concurrently.
+- _"How are you?"_ (no research) — quick reply, no job group.
+- **Click Cancel on an in-flight card** — the cancellation routes
+ through, the peers' tasks raise `CancelledError`, and their responses
+ come back as `cancelled`.
+
+## Requirements
+
+- `OPENAI_API_KEY`
+- `DEEPGRAM_API_KEY`
+- `CARTESIA_API_KEY`
+
+A `.env` in the example folder is the easiest way to set these (see
+`examples/multi-worker/env.example`).
+
+## What this example _doesn't_ show
+
+Real worker integrations (the peers are simulated), LLM-driven peers
+(these are pure data-fetch — a peer can itself be an `LLMWorker`),
+streaming chunks (`send_job_stream_data` for progressive output), or
+worker-to-worker fan-out (nested job groups).
diff --git a/examples/multi-worker/ui-worker/async-tasks/bot.py b/examples/multi-worker/ui-worker/async-tasks/bot.py
new file mode 100644
index 000000000..dc71113bc
--- /dev/null
+++ b/examples/multi-worker/ui-worker/async-tasks/bot.py
@@ -0,0 +1,378 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Async tasks — the UIWorker fans out long-running work and streams progress.
+
+The user asks the assistant to research a topic. The UIWorker dispatches
+three peer workers (Wikipedia, news, scholarly papers) in parallel via
+``start_user_job_group``. Each worker emits progress updates while it
+works. ``UIWorker`` forwards every lifecycle event to the client as
+``ui-task`` envelopes (``group_started``, ``task_update``,
+``task_completed``, ``group_completed``), which the client renders as
+in-flight cards with per-worker status. The user can cancel a group
+mid-flight via ``client.cancelUITask(task_id)``, which sends a reserved
+``__cancel_task`` event that the worker turns into a ``cancel_job_group``
+call.
+
+Architecture::
+
+ Main worker (PipelineWorker, owns transport + RTVI):
+ transport.in → STT → user_agg → LLM → TTS → transport.out → assistant_agg
+ └── answer_about_screen(query) tool
+ └── params.pipeline_worker.job("ui", name="respond", payload={query})
+
+ ResearchWorker (UIWorker):
+ └── @tool reply(answer, research_query=None)
+ └── (if research_query) start_user_job_group("wikipedia", "news", "scholar")
+
+ Three peer workers (BaseWorker each):
+ WikipediaResearcher · NewsResearcher · ScholarResearcher
+
+The workers are deliberately simulated with ``asyncio.sleep`` and canned
+summaries so the demo focuses on the protocol, not the AI. A real app
+would wire each worker to its own data source.
+
+``start_user_job_group`` dispatches the group on a background task and
+returns immediately, so the spoken "researching X" acknowledgement frees
+the main LLM to take new turns while the workers continue.
+
+Run::
+
+ uv run python bot.py
+
+Then open the client at ``http://localhost:5173`` (see ``README.md``).
+
+Requirements:
+
+- OPENAI_API_KEY
+- DEEPGRAM_API_KEY
+- CARTESIA_API_KEY
+"""
+
+import asyncio
+import os
+import random
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.bus.messages import BusJobRequestMessage
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.base_worker import BaseWorker
+from pipecat.pipeline.job_context import JobError
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+ LLMContextAggregatorPair,
+ LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.workers.llm import tool
+from pipecat.workers.ui import UIWorker
+
+load_dotenv(override=True)
+
+MAIN_NAME = "main"
+
+transport_params = {
+ "daily": lambda: DailyParams(audio_in_enabled=True, audio_out_enabled=True),
+ "webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True),
+}
+
+
+VOICE_PROMPT = """\
+You are the voice layer of a research assistant. A separate UI \
+layer sees the page and dispatches research tasks.
+
+For every user utterance involving research (asking about a topic, \
+launching a search, asking for follow-ups), call \
+``answer_about_screen`` with the user's request verbatim. The \
+tool's response is the spoken reply, already TTS-ready.
+
+Only respond directly for pure pleasantries (greetings, thanks, \
+goodbyes). Keep direct replies to one short spoken sentence."""
+
+
+# The UI wire-format guide (UI_STATE_PROMPT_GUIDE) is appended to the LLM's
+# system instruction automatically by UIWorker, so this prompt only needs the
+# app-specific behavior.
+UI_PROMPT = """\
+You help the user research topics. When the user names something \
+to look up, kick off a parallel research task across three worker \
+sources (Wikipedia, news, scholarly papers).
+
+## Tool: reply
+
+Every turn calls ``reply`` exactly once. One tool call per turn.
+
+``reply(answer, research_query=None)``:
+
+- ``answer`` (REQUIRED): the spoken reply, plain language, one \
+short sentence. No markdown, no symbols.
+- ``research_query`` (OPTIONAL): the topic to research. When set, \
+the server fans out three worker agents in parallel and streams \
+their progress to an in-flight panel on the page. The workers run \
+in the background; you do NOT wait for results. Just speak a brief \
+acknowledgement.
+
+## Decision rules
+
+- **User asks to research / look up / find out about something** → \
+set ``research_query`` to the topic and answer with a brief \
+acknowledgement ("Researching the Mariana Trench now"). The server \
+handles the rest; results stream onto the page.
+- **User asks a quick question you can answer immediately** → just \
+``answer``. Don't kick off a research task for trivia or for \
+questions about the in-flight tasks themselves.
+- **User asks about ongoing research** → just ``answer`` (the \
+results panel on screen shows progress).
+
+## Examples
+
+- "Research the Mariana Trench." → \
+``reply(answer="Researching the Mariana Trench now.", research_query="Mariana Trench")``
+- "Look up octopus cognition." → \
+``reply(answer="Looking that up.", research_query="octopus cognition")``
+- "How many neurons does an octopus have?" (quick question, no \
+research needed) → ``reply(answer="About five hundred million.")``
+- "Hi." → ``reply(answer="Hi! What would you like to research?")``"""
+
+
+class _SimulatedResearcher(BaseWorker):
+ """BaseWorker peer that fakes a research task with progress updates.
+
+ Receives a ``payload={"query": ...}``. Emits a few ``send_job_update``
+ messages with progress text, then a final ``send_job_response``
+ carrying a canned summary. The randomized ``asyncio.sleep`` makes the
+ workers feel like they run at different paces, which shows off the
+ streaming UI.
+
+ Subclasses set ``source_name`` and provide ``summarize(query)``.
+ """
+
+ source_name: str = "researcher"
+
+ def summarize(self, query: str) -> str:
+ return f"Generic results for '{query}'."
+
+ async def on_job_request(self, message: BusJobRequestMessage) -> None:
+ await super().on_job_request(message)
+ job_id = message.job_id
+ query = (message.payload or {}).get("query", "")
+ try:
+ await asyncio.sleep(random.uniform(0.4, 1.2))
+ await self.send_job_update(job_id, {"text": f"searching {self.source_name}…"})
+
+ await asyncio.sleep(random.uniform(0.6, 1.4))
+ n = random.randint(3, 8)
+ await self.send_job_update(job_id, {"text": f"found {n} results"})
+
+ await asyncio.sleep(random.uniform(0.5, 1.5))
+ await self.send_job_update(job_id, {"text": "summarizing"})
+
+ await asyncio.sleep(random.uniform(0.4, 0.9))
+ await self.send_job_response(job_id, response={"summary": self.summarize(query)})
+ except asyncio.CancelledError:
+ # The base worker's cancellation hook auto-emits a CANCELLED
+ # response; just bail.
+ raise
+
+
+class WikipediaResearcher(_SimulatedResearcher):
+ source_name = "wikipedia"
+
+ def summarize(self, query: str) -> str:
+ return (
+ f"Wikipedia overview of {query}: a one-paragraph summary covering "
+ "the historical background, key facts, and major figures."
+ )
+
+
+class NewsResearcher(_SimulatedResearcher):
+ source_name = "news"
+
+ def summarize(self, query: str) -> str:
+ return (
+ f"Recent news on {query}: three headlines from the past month, "
+ "a short context paragraph, and any active developments."
+ )
+
+
+class ScholarResearcher(_SimulatedResearcher):
+ source_name = "scholar"
+
+ def summarize(self, query: str) -> str:
+ return (
+ f"Scholarly take on {query}: two highly cited papers, the "
+ "consensus position, and a notable debate or open question."
+ )
+
+
+class ResearchWorker(UIWorker):
+ """UIWorker that kicks off background research job groups.
+
+ The custom ``@tool reply`` has a ``research_query`` field. When the
+ LLM sets it, the tool fires ``start_user_job_group(...)`` against the
+ three peer workers — fire-and-forget from the LLM's perspective, so
+ the tool returns immediately with the spoken acknowledgement. The
+ ``UIWorker`` forwards every job lifecycle event to the client as
+ ``ui-task`` envelopes, where the client renders progress and a cancel
+ button.
+ """
+
+ def __init__(self):
+ llm = OpenAILLMService(
+ api_key=os.environ["OPENAI_API_KEY"],
+ settings=OpenAILLMService.Settings(system_instruction=UI_PROMPT),
+ )
+ super().__init__("ui", llm=llm)
+
+ @tool
+ async def reply(
+ self,
+ params: FunctionCallParams,
+ answer: str,
+ research_query: str | None = None,
+ ):
+ """Reply to the user. Optionally kick off background research.
+
+ Always called exactly once per turn. ``answer`` is required.
+
+ Args:
+ answer: The spoken reply in plain language. One short
+ sentence. For research turns, a brief acknowledgement
+ like "Researching X now."
+ research_query: Optional topic to research. When set, the
+ server fans out three worker agents in parallel and
+ streams progress to the page. Workers run in the
+ background; the LLM does NOT wait for results.
+ """
+ logger.info(f"{self}: reply(answer={answer!r}, research_query={research_query!r})")
+ if research_query:
+ await self.start_user_job_group(
+ "wikipedia",
+ "news",
+ "scholar",
+ payload={"query": research_query},
+ label=f"Research: {research_query}",
+ )
+ await self.respond_to_job(speak=answer)
+ await params.result_callback(None)
+
+
+async def answer_about_screen(params: FunctionCallParams, query: str):
+ """Forward the user's request to the screen-aware research worker.
+
+ Args:
+ query (str): The user's request, passed verbatim.
+ """
+ logger.info(f"answer_about_screen('{query}')")
+ try:
+ async with params.pipeline_worker.job(
+ "ui", name="respond", payload={"query": query}, timeout=10
+ ) as t:
+ pass
+ except JobError as e:
+ logger.warning(f"ui job failed: {e}")
+ await params.result_callback("Something went wrong on my side.")
+ return
+
+ speak = (t.response or {}).get("speak")
+ await params.result_callback(speak or "I'm not sure how to answer that.")
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+ logger.info("Starting async-tasks bot")
+
+ runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+ stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
+ tts = CartesiaTTSService(
+ api_key=os.environ["CARTESIA_API_KEY"],
+ settings=CartesiaTTSService.Settings(
+ voice=os.getenv("CARTESIA_VOICE_ID", "71a7ad14-091c-4e8e-a314-022ece01c121"),
+ ),
+ )
+ llm = OpenAILLMService(
+ api_key=os.environ["OPENAI_API_KEY"],
+ settings=OpenAILLMService.Settings(system_instruction=VOICE_PROMPT),
+ )
+ llm.register_direct_function(answer_about_screen, cancel_on_interruption=False, timeout_secs=30)
+
+ context = LLMContext(tools=ToolsSchema(standard_tools=[answer_about_screen]))
+ aggregators = LLMContextAggregatorPair(
+ context,
+ user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+ )
+
+ pipeline = Pipeline(
+ [
+ transport.input(),
+ stt,
+ aggregators.user(),
+ llm,
+ tts,
+ transport.output(),
+ aggregators.assistant(),
+ ]
+ )
+
+ worker = PipelineWorker(
+ pipeline,
+ name=MAIN_NAME,
+ params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
+ idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+ )
+
+ @transport.event_handler("on_client_connected")
+ async def on_client_connected(transport, client):
+ logger.info("Client connected")
+ context.add_message(
+ {
+ "role": "developer",
+ "content": (
+ "Greet the user briefly. Tell them they can ask you to "
+ "research any topic. One short sentence."
+ ),
+ }
+ )
+ await worker.queue_frame(LLMRunFrame())
+
+ @transport.event_handler("on_client_disconnected")
+ async def on_client_disconnected(transport, client):
+ logger.info("Client disconnected")
+ await runner.cancel()
+
+ await runner.launch_worker(ResearchWorker())
+ await runner.launch_worker(WikipediaResearcher("wikipedia"))
+ await runner.launch_worker(NewsResearcher("news"))
+ await runner.launch_worker(ScholarResearcher("scholar"))
+ await runner.launch_worker(worker)
+
+ await runner.run()
+
+
+async def bot(runner_args: RunnerArguments):
+ """Main bot entry point compatible with Pipecat Cloud."""
+ transport = await create_transport(runner_args, transport_params)
+ await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+ from pipecat.runner.run import main
+
+ main()
diff --git a/examples/multi-worker/ui-worker/async-tasks/client/index.html b/examples/multi-worker/ui-worker/async-tasks/client/index.html
new file mode 100644
index 000000000..a38b82699
--- /dev/null
+++ b/examples/multi-worker/ui-worker/async-tasks/client/index.html
@@ -0,0 +1,45 @@
+
+
+
+
+
+ Async tasks — UIAgent demo
+
+
+
+
+
Research lab
+
+
+
+
+
+ Try: "research the Mariana Trench" — the agent fans out three
+ worker researchers (Wikipedia, news, scholarly papers) in
+ parallel. Each streams progress updates while it works. You
+ can cancel any group mid-flight.
+
+
+
+
In flight
+
+ No tasks yet. Ask the assistant to research something.
+