From 0ff70c222d0f2f18dd58926e26584fd360a55048 Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Fri, 19 Jun 2026 18:21:28 +0200 Subject: [PATCH] feat(streaming): surface native web-search activity per chunk Add a live tool_activity signal on TextChunk so consumers can render web-search progress while it happens, instead of only seeing the final grounding on the output. - types: ToolActivity model + ToolActivityStatus (started/completed) - TextChunk gains tool_activity; base Stream._parse_chunk parses and threads it like reasoning, and keeps activity-only chunks - OpenResponses (OpenAI + xAI): web_search_call.in_progress -> started, .completed -> completed - Anthropic: server_tool_use(web_search) -> started, web_search_tool_result -> completed Streaming-only by design: final results already live on TextOutput.grounding. Gemini has no in-stream signal (terminal-only). --- src/celeste/modalities/text/io.py | 2 ++ .../protocols/openresponses/streaming.py | 16 ++++++++++++++++ .../providers/anthropic/messages/streaming.py | 19 +++++++++++++++++++ src/celeste/streaming.py | 12 +++++++++++- src/celeste/types.py | 16 ++++++++++++++++ 5 files changed, 64 insertions(+), 1 deletion(-) diff --git a/src/celeste/modalities/text/io.py b/src/celeste/modalities/text/io.py index dee073f..4e69ef4 100644 --- a/src/celeste/modalities/text/io.py +++ b/src/celeste/modalities/text/io.py @@ -21,6 +21,7 @@ Message, Role, TextContent, + ToolActivity, VideoContent, ) @@ -78,6 +79,7 @@ class TextChunk(Chunk[str]): """Chunk for text streaming.""" reasoning: str | None = None + tool_activity: ToolActivity | None = None finish_reason: TextFinishReason | None = None usage: TextUsage | None = None diff --git a/src/celeste/protocols/openresponses/streaming.py b/src/celeste/protocols/openresponses/streaming.py index 8df3ef3..59e23fb 100644 --- a/src/celeste/protocols/openresponses/streaming.py +++ b/src/celeste/protocols/openresponses/streaming.py @@ -3,6 +3,7 @@ from typing import Any, ClassVar from celeste.io import FinishReason +from celeste.types import ToolActivity, ToolActivityStatus from .client import OpenResponsesClient @@ -47,6 +48,21 @@ def _parse_chunk_reasoning(self, event_data: dict[str, Any]) -> str | None: return event_data.get("delta") or None return None + def _parse_chunk_tool_activity( + self, event_data: dict[str, Any] + ) -> ToolActivity | None: + """Extract native web-search activity from SSE event.""" + event_type = event_data.get("type") + if event_type == "response.web_search_call.in_progress": + return ToolActivity( + tool_name="web_search", status=ToolActivityStatus.STARTED + ) + if event_type == "response.web_search_call.completed": + return ToolActivity( + tool_name="web_search", status=ToolActivityStatus.COMPLETED + ) + return None + def _parse_chunk_usage( self, event_data: dict[str, Any] ) -> dict[str, int | float | None] | None: diff --git a/src/celeste/providers/anthropic/messages/streaming.py b/src/celeste/providers/anthropic/messages/streaming.py index 7da9695..b29d5e9 100644 --- a/src/celeste/providers/anthropic/messages/streaming.py +++ b/src/celeste/providers/anthropic/messages/streaming.py @@ -5,6 +5,7 @@ from typing import Any from celeste.io import FinishReason +from celeste.types import ToolActivity, ToolActivityStatus from .client import AnthropicMessagesClient @@ -110,6 +111,24 @@ def _parse_chunk_reasoning(self, event_data: dict[str, Any]) -> str | None: return delta.get("thinking") or None return None + def _parse_chunk_tool_activity( + self, event_data: dict[str, Any] + ) -> ToolActivity | None: + """Extract native web-search activity from content block starts.""" + if event_data.get("type") != "content_block_start": + return None + block = event_data.get("content_block", {}) + block_type = block.get("type") + if block_type == "server_tool_use" and block.get("name") == "web_search": + return ToolActivity( + tool_name="web_search", status=ToolActivityStatus.STARTED + ) + if block_type == "web_search_tool_result": + return ToolActivity( + tool_name="web_search", status=ToolActivityStatus.COMPLETED + ) + return None + def _parse_chunk_usage( self, event_data: dict[str, Any] ) -> dict[str, int | float | None] | None: diff --git a/src/celeste/streaming.py b/src/celeste/streaming.py index e0ab5a0..67b6cfa 100644 --- a/src/celeste/streaming.py +++ b/src/celeste/streaming.py @@ -15,7 +15,7 @@ from celeste.io import FinishReason, Output, Usage from celeste.parameters import Parameters from celeste.tools import ToolCall, validate_tool_calls -from celeste.types import RawUsage +from celeste.types import RawUsage, ToolActivity async def enrich_stream_errors( @@ -114,6 +114,12 @@ def _parse_chunk_reasoning(self, event_data: dict[str, Any]) -> str | None: """Parse reasoning delta from chunk event. Override in providers with reasoning.""" return None + def _parse_chunk_tool_activity( + self, event_data: dict[str, Any] + ) -> ToolActivity | None: + """Parse native tool activity from chunk event. Override in providers that emit it.""" + return None + def _wrap_chunk_content(self, raw_content: Any) -> Any: # noqa: ANN401 """Wrap raw content into chunk content type. Override for type transformation.""" return raw_content @@ -130,11 +136,13 @@ def _parse_chunk(self, event: dict[str, Any]) -> Chunk | None: ) content = self._parse_chunk_content(event) reasoning = self._parse_chunk_reasoning(event) + tool_activity = self._parse_chunk_tool_activity(event) usage = self._get_chunk_usage(event) finish_reason = self._get_chunk_finish_reason(event) if ( content is None and reasoning is None + and tool_activity is None and usage is None and finish_reason is None ): @@ -147,6 +155,8 @@ def _parse_chunk(self, event: dict[str, Any]) -> Chunk | None: kwargs: dict[str, Any] = {} if reasoning is not None: kwargs["reasoning"] = reasoning + if tool_activity is not None: + kwargs["tool_activity"] = tool_activity return self._chunk_class( # type: ignore[return-value] content=content, finish_reason=finish_reason, diff --git a/src/celeste/types.py b/src/celeste/types.py index f044bc0..72aca99 100644 --- a/src/celeste/types.py +++ b/src/celeste/types.py @@ -89,6 +89,20 @@ class ToolCall(BaseModel): arguments: dict[str, Any] +class ToolActivityStatus(StrEnum): + """Lifecycle of a native (server-side) tool invocation during streaming.""" + + STARTED = "started" + COMPLETED = "completed" + + +class ToolActivity(BaseModel): + """A native tool invocation surfaced live in a streaming chunk.""" + + tool_name: str + status: ToolActivityStatus + + class Message(BaseModel): """A message in a conversation.""" @@ -117,6 +131,8 @@ class Message(BaseModel): "Role", "TextContent", "TextPart", + "ToolActivity", + "ToolActivityStatus", "ToolCall", "ToolResultContent", "VideoContent",