feat(mcp): add nonblocking_tools to MCPToolset

examples/voice_agents/mcp/mcp-agent.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import datetime
 
 from dotenv import load_dotenv
 
@@ -15,8 +16,14 @@ class MyAgent(Agent):
     def __init__(self) -> None:
         super().__init__(
             instructions=(
-                "You can retrieve data via the MCP server. The interface is voice-based: "
-                "accept spoken user queries and respond with synthesized speech."
+                "You are a friendly travel assistant that communicates via voice. "
+                "Avoid emojis and markdown — speak naturally and concisely. "
+                "Use the get_weather tool when the user asks about the weather somewhere. "
+                "Use the book_flight tool when the user wants to book a flight; it takes "
+                "a couple of minutes and will report progress along the way — narrate the "
+                "updates naturally and don't make up flight details. "
+                "Don't repeat information you have already said in the conversation. "
+                f"Today is {datetime.now().strftime('%Y-%m-%d %A')}"
             ),
         )
 
@@ -39,7 +46,16 @@ async def entrypoint(ctx: JobContext):
         turn_detection=MultilingualModel(),
         tools=[
             mcp.MCPToolset(
-                id="mcp_toolset_1", mcp_server=mcp.MCPServerHTTP(url="http://localhost:8000/sse")
+                id="mcp_toolset_1",
+                mcp_server=mcp.MCPServerHTTP(
+                    url="http://localhost:8000/sse",
+                    # book_flight takes ~70s; bump the per-request timeout above that.
+                    client_session_timeout_seconds=120,
+                ),
+                # Only book_flight is long-running; get_weather can block the reply.
+                # Pass `True` (the default) to make every MCP tool non-blocking, or
+                # `False` to fall back to fully blocking execution.
+                nonblocking_tools=["book_flight"],
             )
         ],
     )

examples/voice_agents/mcp/server.py

@@ -1,12 +1,51 @@
-from mcp.server.fastmcp import FastMCP
+import asyncio
+import random
+
+from mcp.server.fastmcp import Context, FastMCP
 
 mcp = FastMCP("Demo 🚀")
 
 
 @mcp.tool()
 def get_weather(location: str) -> str:
+    """Get the current weather for a location."""
     return f"The weather in {location} is a perfect sunny 70°F today. Enjoy your day!"
 
 
+@mcp.tool()
+async def book_flight(origin: str, destination: str, date: str, ctx: Context) -> str:
+    """Book a flight. Takes a couple of minutes — emits progress notifications
+    while searching airlines and confirming the booking."""
+    await ctx.report_progress(
+        0.0,
+        1.0,
+        f"Searching flights from {origin} to {destination} on {date}. "
+        "This will take a couple of minutes.",
+    )
+
+    # Phase 1: searching airlines
+    await asyncio.sleep(30)
+
+    airlines = random.sample(["United", "Delta", "American", "JetBlue", "Southwest", "Alaska"], k=3)
+    prices = {a: random.randint(180, 650) for a in airlines}
+    cheapest = min(prices, key=lambda a: prices[a])
+
+    await ctx.report_progress(
+        0.5,
+        1.0,
+        f"Found {len(airlines)} options. Best price: ${prices[cheapest]} on {cheapest}. "
+        "Confirming the booking now.",
+    )
+
+    # Phase 2: confirming booking
+    await asyncio.sleep(40)
+
+    confirmation = f"FL-{random.randint(100000, 999999)}"
+    return (
+        f"Flight booked! {cheapest} from {origin} to {destination} on {date}. "
+        f"Price: ${prices[cheapest]}. Confirmation: {confirmation}."
+    )
+
+
 if __name__ == "__main__":
     mcp.run(transport="sse")

livekit-agents/livekit/agents/llm/async_toolset.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import copy
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal, get_origin, get_type_hints
 
@@ -185,14 +186,16 @@ def __init__(
             self._wrap_tool(t) if isinstance(t, FunctionTool | RawFunctionTool) else t
             for t in self._tools
         ]
-        self._tools.extend([get_running_tasks, cancel_task])
-
         self._running_tasks: dict[str, _RunningTask] = {}
 
         # speech delivery — shared across all tools in this toolset
         self._pending_updates: list[_PendingUpdate] = []
         self._reply_task: asyncio.Task[None] | None = None
 
+    @property
+    def tools(self) -> Sequence[Tool | Toolset]:
+        return [*self._tools, get_running_tasks, cancel_task]
+
     async def cancel(self, call_id: str) -> bool:
         task = self._running_tasks.get(call_id)
         if task is not None:

livekit-agents/livekit/agents/llm/mcp.py

@@ -5,7 +5,7 @@
 import asyncio
 import json
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
 from datetime import timedelta
@@ -17,7 +17,8 @@
 from typing_extensions import Self
 
 from ..log import logger
-from .tool_context import Toolset
+from .async_toolset import AsyncRunContext, AsyncToolset, cancel_task, get_running_tasks
+from .tool_context import Tool, Toolset
 
 try:
     import httpx
@@ -87,6 +88,7 @@ def __init__(
 
         self._cache_dirty = True
         self._lk_tools: list[MCPTool] | None = None
+        self._lk_tools_nonblocking: bool | frozenset[str] = False
 
         self._client_task: asyncio.Task[None] | None = None
         self._closing_ev = asyncio.Event()
@@ -140,20 +142,38 @@ async def _run_client(self, ready_fut: asyncio.Future[None]) -> None:
             self._lk_tools = None
             self._closing_ev.clear()
 
-    async def list_tools(self) -> list[MCPTool]:
+    async def list_tools(
+        self, *, nonblocking_tools: bool | frozenset[str] = False
+    ) -> list[MCPTool]:
         if self._client is None:
             raise RuntimeError("MCPServer isn't initialized")
 
-        if not self._cache_dirty and self._lk_tools is not None:
+        if (
+            not self._cache_dirty
+            and self._lk_tools is not None
+            and self._lk_tools_nonblocking == nonblocking_tools
+        ):
             return self._lk_tools
 
+        def _is_nonblocking(name: str) -> bool:
+            if isinstance(nonblocking_tools, bool):
+                return nonblocking_tools
+            return name in nonblocking_tools
+
         tools = await self._client.list_tools()
         lk_tools = [
-            self._make_function_tool(tool.name, tool.description, tool.inputSchema, tool.meta)
+            self._make_function_tool(
+                tool.name,
+                tool.description,
+                tool.inputSchema,
+                tool.meta,
+                async_mode=_is_nonblocking(tool.name),
+            )
             for tool in tools.tools
         ]
 
         self._lk_tools = lk_tools
+        self._lk_tools_nonblocking = nonblocking_tools
         self._cache_dirty = False
         return lk_tools
 
@@ -163,17 +183,12 @@ def _make_function_tool(
         description: str | None,
         input_schema: dict[str, Any],
         meta: dict[str, Any] | None,
+        *,
+        async_mode: bool = False,
     ) -> MCPTool:
-        async def _tool_called(raw_arguments: dict[str, Any]) -> Any:
-            # In case (somehow), the tool is called after the MCPServer aclose.
-            if self._client is None:
-                raise ToolError(
-                    "Tool invocation failed: internal service is unavailable. "
-                    "Please check that the MCPServer is still running."
-                )
-
-            tool_result = await self._client.call_tool(name, raw_arguments)
-
+        async def _resolve(
+            tool_result: mcp.types.CallToolResult, raw_arguments: dict[str, Any]
+        ) -> Any:
             if tool_result.isError:
                 error_str = "\n".join(
                     part.text if hasattr(part, "text") else str(part)
@@ -187,6 +202,53 @@ async def _tool_called(raw_arguments: dict[str, Any]) -> Any:
                 resolved = await resolved
             return resolved
 
+        if async_mode:
+
+            async def _tool_called_async(
+                ctx: AsyncRunContext, raw_arguments: dict[str, Any]
+            ) -> Any:
+                if self._client is None:
+                    raise ToolError(
+                        "Tool invocation failed: internal service is unavailable. "
+                        "Please check that the MCPServer is still running."
+                    )
+
+                async def _on_progress(
+                    progress: float, total: float | None, message: str | None
+                ) -> None:
+                    if message:
+                        logger.debug(
+                            "MCPTool progress callback",
+                            extra={
+                                "tool_name": name,
+                                "progress": progress,
+                                "total": total,
+                                "progress_message": message,
+                            },
+                        )
+                        await ctx.update(message)
+
+                tool_result = await self._client.call_tool(
+                    name, raw_arguments, progress_callback=_on_progress
+                )
+                return await _resolve(tool_result, raw_arguments)
+
+            impl: Callable[..., Awaitable[Any]] = _tool_called_async
+        else:
+
+            async def _tool_called(raw_arguments: dict[str, Any]) -> Any:
+                # in case the tool is called after the MCPServer aclose.
+                if self._client is None:
+                    raise ToolError(
+                        "Tool invocation failed: internal service is unavailable. "
+                        "Please check that the MCPServer is still running."
+                    )
+
+                tool_result = await self._client.call_tool(name, raw_arguments)
+                return await _resolve(tool_result, raw_arguments)
+
+            impl = _tool_called
+
         raw_schema = {
             "name": name,
             "description": description,
@@ -195,7 +257,7 @@ async def _tool_called(raw_arguments: dict[str, Any]) -> Any:
         if meta:
             raw_schema["meta"] = meta
 
-        return function_tool(_tool_called, raw_schema=raw_schema)
+        return function_tool(impl, raw_schema=raw_schema)
 
     async def aclose(self) -> None:
         self._closing_ev.set()
@@ -353,11 +415,13 @@ async def _streamable_http_with_client():  # type: ignore[no-untyped-def]
                 httpx_client_factory=self._create_http_client,
             )
 
-    async def list_tools(self) -> list[MCPTool]:
+    async def list_tools(
+        self, *, nonblocking_tools: bool | frozenset[str] = False
+    ) -> list[MCPTool]:
         """
         List tools from the MCP server, filtered by allowed_tools if specified.
         """
-        all_tools = await super().list_tools()
+        all_tools = await super().list_tools(nonblocking_tools=nonblocking_tools)
 
         # If no filter is set, return all tools
         if self._allowed_tools is None:
@@ -431,20 +495,53 @@ def __repr__(self) -> str:
         return f"MCPServerStdio(command={self.command}, args={self.args}, cwd={self.cwd})"
 
 
-class MCPToolset(Toolset):
+class MCPToolset(AsyncToolset):
     """A toolset that exposes tools from a Model Context Protocol (MCP) server.
 
     MCPToolset wraps an ``MCPServer`` instance and makes its tools available for
     use by an ``Agent``. On ``setup()``, it connects to the MCP server (if not
     already connected), fetches the available tools, and caches them locally.
+
+    The ``nonblocking_tools`` parameter controls which MCP tools run as
+    background tasks (with :class:`AsyncRunContext` so MCP progress
+    notifications are forwarded via ``ctx.update()``) versus blocking the
+    agent's reply loop until completion:
+
+    * ``True`` (default) — all MCP tools are non-blocking.
+    * ``False`` — all MCP tools block.
+    * ``list[str]`` — only the named tools are non-blocking; the rest block.
+
+    When at least one tool is non-blocking, the toolset also exposes
+    ``get_running_tasks`` and ``cancel_task`` to the LLM.
     """
 
-    def __init__(self, *, id: str, mcp_server: MCPServer) -> None:
-        super().__init__(id=id)
+    def __init__(
+        self,
+        *,
+        id: str,
+        mcp_server: MCPServer,
+        nonblocking_tools: bool | list[str] = True,
+        on_duplicate_call: AsyncToolset.DuplicateMode = "confirm",
+    ) -> None:
+        super().__init__(id=id, on_duplicate_call=on_duplicate_call)
         self._mcp_server = mcp_server
+        self._nonblocking_tools: bool | frozenset[str] = (
+            nonblocking_tools
+            if isinstance(nonblocking_tools, bool)
+            else frozenset(nonblocking_tools)
+        )
         self._initialized = False
         self._lock = asyncio.Lock()
 
+    @property
+    def tools(self) -> Sequence[Tool | Toolset]:
+        # Only expose the AsyncToolset helper tools when at least one MCP tool
+        # runs non-blocking; otherwise the LLM would see `get_running_tasks` /
+        # `cancel_task` without anything to manage.
+        if self._nonblocking_tools:
+            return [*self._tools, get_running_tasks, cancel_task]
+        return list(self._tools)
+
     async def setup(self, *, reload: bool = False) -> Self:
         """Initialize the MCP server connection and fetch available tools.
 
@@ -466,8 +563,8 @@ async def setup(self, *, reload: bool = False) -> Self:
             elif reload:
                 self._mcp_server.invalidate_cache()
 
-            tools = await self._mcp_server.list_tools()
-            self._tools = tools
+            tools = await self._mcp_server.list_tools(nonblocking_tools=self._nonblocking_tools)
+            self._tools = [self._wrap_tool(tool) for tool in tools]
             self._initialized = True
             return self