feat: add provider tool call lifecycle events

livekit-agents/livekit/agents/llm/__init__.py

@@ -23,6 +23,7 @@
     FunctionToolCall,
     LLMError,
     LLMStream,
+    ProviderToolCall,
 )
 from .realtime import (
     GenerationCreatedEvent,
@@ -104,6 +105,7 @@
     "utils",
     "remote_chat_context",
     "FunctionToolCall",
+    "ProviderToolCall",
     "RealtimeModel",
     "RealtimeError",
     "RealtimeModelError",

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

@@ -76,6 +76,26 @@ class ChoiceDelta(BaseModel):
     """Provider-specific extra data (e.g., Google thought signatures)."""
 
 
+class ProviderToolCall(BaseModel):
+    """A provider-executed tool call, surfaced by a plugin as it runs.
+
+    These tools run inside the model provider's own stream (e.g. Mistral web search,
+    xAI WebSearch). A plugin emits the ``provider_tool_call`` event on the
+    :class:`LLM` (an ``EventEmitter``, like ``metrics_collected``) with ``phase="started"``
+    when the provider begins the call and ``phase="done"`` when it finishes; the framework
+    bridges these to ``provider_tool_execution_updated`` session events. They are never
+    executed locally.
+    """
+
+    type: Literal["provider_tool_call"] = "provider_tool_call"
+    phase: Literal["started", "done"]
+    call_id: str
+    name: str
+    arguments: str = ""
+    result: str | None = None
+    """Tool result, populated on ``phase="done"`` when the provider returns one."""
+
+
 class ChatChunk(BaseModel):
     id: str
     delta: ChoiceDelta | None = None
@@ -105,7 +125,7 @@ class LLMError(BaseModel):
 
 class LLM(
     ABC,
-    rtc.EventEmitter[Literal["metrics_collected", "error"] | TEvent],
+    rtc.EventEmitter[Literal["metrics_collected", "error", "provider_tool_call"] | TEvent],
     Generic[TEvent],
 ):
     def __init__(self) -> None:

livekit-agents/livekit/agents/voice/__init__.py

@@ -19,6 +19,9 @@
     ErrorEvent,
     FunctionToolsExecutedEvent,
     MetricsCollectedEvent,
+    ProviderToolCallEnded,
+    ProviderToolCallStarted,
+    ProviderToolExecutionUpdatedEvent,
     RunContext,
     SessionUsageUpdatedEvent,
     SpeechCreatedEvent,
@@ -80,6 +83,9 @@
     "ToolCallUpdated",
     "ToolCallEnded",
     "ToolReplyUpdated",
+    "ProviderToolExecutionUpdatedEvent",
+    "ProviderToolCallStarted",
+    "ProviderToolCallEnded",
     "UserTurnExceededEvent",
     "KeytermsOptions",
     "KeytermDetectionOptions",

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -59,6 +59,9 @@
     ErrorEvent,
     FunctionToolsExecutedEvent,
     MetricsCollectedEvent,
+    ProviderToolCallEnded,
+    ProviderToolCallStarted,
+    ProviderToolExecutionUpdatedEvent,
     SessionUsageUpdatedEvent,
     SpeechCreatedEvent,
     UserInputTranscribedEvent,
@@ -747,12 +750,14 @@ def _update_models(
             if isinstance(old_llm, llm.LLM):
                 old_llm.off("metrics_collected", self._on_metrics_collected)
                 old_llm.off("error", self._on_error)
+                old_llm.off("provider_tool_call", self._on_provider_tool_call)
 
             self._agent._llm = new_llm  # llm_node reads activity.llm per generation
             if isinstance(self.llm, llm.LLM):
                 self.llm.prewarm()
                 self.llm.on("metrics_collected", self._on_metrics_collected)
                 self.llm.on("error", self._on_error)
+                self.llm.on("provider_tool_call", self._on_provider_tool_call)
 
         if is_given(new_tts):
             old_tts = self.tts
@@ -998,6 +1003,7 @@ async def _start_session(self, *, reuse_resources: _ReusableResources | None = N
         if isinstance(self.llm, llm.LLM):
             self.llm.on("metrics_collected", self._on_metrics_collected)
             self.llm.on("error", self._on_error)
+            self.llm.on("provider_tool_call", self._on_provider_tool_call)
 
         if isinstance(self.stt, stt.STT):
             self.stt.on("metrics_collected", self._on_metrics_collected)
@@ -1298,6 +1304,7 @@ async def _close_session(self) -> None:
         if isinstance(self.llm, llm.LLM):
             self.llm.off("metrics_collected", self._on_metrics_collected)
             self.llm.off("error", self._on_error)
+            self.llm.off("provider_tool_call", self._on_provider_tool_call)
 
         if isinstance(self.llm, llm.RealtimeModel) and self._rt_session is not None:
             self._rt_session.off("generation_created", self._on_generation_created)
@@ -1865,6 +1872,26 @@ async def _wait_for_eou() -> None:
 
     # -- Realtime Session events --
 
+    def _on_provider_tool_call(self, call: llm.ProviderToolCall) -> None:
+        # bridge the LLM's provider-tool lifecycle onto the session, parallel to
+        # `tool_execution_updated` for locally-executed tools
+        update: ProviderToolCallStarted | ProviderToolCallEnded
+        if call.phase == "started":
+            update = ProviderToolCallStarted(
+                call_id=call.call_id, name=call.name, arguments=call.arguments
+            )
+        else:
+            update = ProviderToolCallEnded(
+                call_id=call.call_id,
+                name=call.name,
+                arguments=call.arguments,
+                result=call.result,
+            )
+        self._session.emit(
+            "provider_tool_execution_updated",
+            ProviderToolExecutionUpdatedEvent(update=update),
+        )
+
     def _on_metrics_collected(
         self,
         ev: STTMetrics | TTSMetrics | VADMetrics | LLMMetrics | RealtimeModelMetrics,

livekit-agents/livekit/agents/voice/events.py

@@ -298,6 +298,7 @@ def _make_update_pair(
     "session_usage_updated",
     "speech_created",
     "tool_execution_updated",
+    "provider_tool_execution_updated",
     "error",
     "close",
     "debug_message",
@@ -526,6 +527,42 @@ class ToolExecutionUpdatedEvent(BaseModel):
     created_at: float = Field(default_factory=time.time)
 
 
+class ProviderToolCallStarted(BaseModel):
+    """A provider-executed (server-side) tool call began running."""
+
+    type: Literal["provider_tool_call_started"] = "provider_tool_call_started"
+    call_id: str
+    name: str
+    arguments: str = ""
+
+
+class ProviderToolCallEnded(BaseModel):
+    """A provider-executed tool call finished, with its result when the provider returns one."""
+
+    type: Literal["provider_tool_call_ended"] = "provider_tool_call_ended"
+    call_id: str
+    name: str
+    arguments: str = ""
+    result: str | None = None
+
+
+class ProviderToolExecutionUpdatedEvent(BaseModel):
+    """One provider-tool lifecycle update, parallel to ``ToolExecutionUpdatedEvent`` for
+    locally-executed tools. Discriminate on ``update.type``: ``provider_tool_call_started``
+    → ``provider_tool_call_ended``.
+
+    Provider tools run inside the model provider's own stream, so they never appear in
+    ``tool_execution_updated`` or ``function_tools_executed``.
+    """
+
+    type: Literal["provider_tool_execution_updated"] = "provider_tool_execution_updated"
+    update: Annotated[
+        ProviderToolCallStarted | ProviderToolCallEnded,
+        Field(discriminator="type"),
+    ]
+    created_at: float = Field(default_factory=time.time)
+
+
 class UserTurnExceededEvent(BaseModel):
     type: Literal["user_turn_exceeded"] = "user_turn_exceeded"
     transcript: str
@@ -592,6 +629,7 @@ class CloseEvent(BaseModel):
     | FunctionToolsExecutedEvent
     | SpeechCreatedEvent
     | ToolExecutionUpdatedEvent
+    | ProviderToolExecutionUpdatedEvent
     | ErrorEvent
     | CloseEvent
     | OverlappingSpeechEvent,

livekit-plugins/livekit-plugins-mistralai/livekit/plugins/mistralai/llm.py

@@ -8,6 +8,7 @@
 from livekit.agents.llm import (
     ChatChunk,
     ChatContext,
+    ProviderToolCall,
     ToolChoice,
 )
 from livekit.agents.types import (
@@ -357,6 +358,15 @@ def _parse_event(
 
         if isinstance(data, ToolExecutionStartedEvent):
             self._provider_tool_args[data.id] = data.arguments
+            self._llm.emit(
+                "provider_tool_call",
+                ProviderToolCall(
+                    phase="started",
+                    call_id=data.id,
+                    name=data.name,
+                    arguments=data.arguments,
+                ),
+            )
 
         elif isinstance(data, ToolExecutionDeltaEvent):
             if data.id not in self._provider_tool_args:
@@ -369,5 +379,15 @@ def _parse_event(
                 "executed provider tool",
                 extra={"function": data.name, "lk.pii.arguments": args, "lk.pii.info": data.info},
             )
+            self._llm.emit(
+                "provider_tool_call",
+                ProviderToolCall(
+                    phase="done",
+                    call_id=data.id,
+                    name=data.name,
+                    arguments=args,
+                    result=str(data.info) if data.info is not None else None,
+                ),
+            )
 
         return chunks

tests/test_plugin_mistralai_llm.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import types
+
+import pytest
+from mistralai.client.models import (
+    ToolExecutionDeltaEvent,
+    ToolExecutionDoneEvent,
+    ToolExecutionStartedEvent,
+)
+
+from livekit.agents import llm
+from livekit.plugins.mistralai.llm import LLMStream
+
+pytestmark = pytest.mark.plugin("mistralai")
+
+
+class _RecordingLLM:
+    """Captures the events the stream emits on its parent LLM."""
+
+    def __init__(self) -> None:
+        self.events: list[tuple[str, object]] = []
+
+    def emit(self, name: str, payload: object) -> None:
+        self.events.append((name, payload))
+
+
+def _stream() -> LLMStream:
+    # bypass __init__/network; _parse_event only needs the arg buffer and the parent LLM
+    stream = LLMStream.__new__(LLMStream)
+    stream._provider_tool_args = {}
+    stream._llm = _RecordingLLM()  # type: ignore[assignment]
+    return stream
+
+
+def _event(data: object) -> object:
+    # _parse_event only reads `ev.data`
+    return types.SimpleNamespace(data=data)
+
+
+class TestProviderToolLifecycle:
+    def test_started_emits_started_event(self) -> None:
+        stream = _stream()
+
+        chunks = stream._parse_event(
+            _event(ToolExecutionStartedEvent(id="t1", name="web_search", arguments='{"q":"x"}')),
+            {},
+        )
+
+        # provider tools flow via the LLM event, not the chunk stream
+        assert chunks == []
+        assert len(stream._llm.events) == 1
+        name, call = stream._llm.events[0]
+        assert name == "provider_tool_call"
+        assert isinstance(call, llm.ProviderToolCall)
+        assert call.phase == "started"
+        assert call.call_id == "t1"
+        assert call.name == "web_search"
+        assert call.arguments == '{"q":"x"}'
+
+    def test_delta_accumulates_without_emitting(self) -> None:
+        stream = _stream()
+        stream._parse_event(
+            _event(ToolExecutionStartedEvent(id="t1", name="web_search", arguments="{")),
+            {},
+        )
+        stream._llm.events.clear()
+
+        chunks = stream._parse_event(
+            _event(ToolExecutionDeltaEvent(id="t1", name="web_search", arguments='"q":"x"}')),
+            {},
+        )
+
+        assert chunks == []
+        assert stream._llm.events == []
+        assert stream._provider_tool_args["t1"] == '{"q":"x"}'
+
+    def test_done_emits_ended_event_with_accumulated_args_and_result(self) -> None:
+        stream = _stream()
+        stream._parse_event(
+            _event(ToolExecutionStartedEvent(id="t1", name="web_search", arguments="{")),
+            {},
+        )
+        stream._parse_event(
+            _event(ToolExecutionDeltaEvent(id="t1", name="web_search", arguments='"q":"x"}')),
+            {},
+        )
+        stream._llm.events.clear()
+
+        stream._parse_event(
+            _event(ToolExecutionDoneEvent(id="t1", name="web_search", info={"answer": 42})),
+            {},
+        )
+
+        assert len(stream._llm.events) == 1
+        name, call = stream._llm.events[0]
+        assert name == "provider_tool_call"
+        assert call.phase == "done"
+        assert call.call_id == "t1"
+        assert call.name == "web_search"
+        assert call.arguments == '{"q":"x"}'
+        assert call.result == str({"answer": 42})
+        # state is popped so a later turn can't leak args
+        assert "t1" not in stream._provider_tool_args
+
+    def test_done_without_start_is_safe(self) -> None:
+        stream = _stream()
+
+        stream._parse_event(
+            _event(ToolExecutionDoneEvent(id="ghost", name="web_search", info=None)),
+            {},
+        )
+
+        assert len(stream._llm.events) == 1
+        _, call = stream._llm.events[0]
+        assert call.phase == "done"
+        assert call.arguments == ""
+        assert call.result is None