From e10d173f51ef03da22cf5f57240dad3777f35a4a Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Tue, 16 Jun 2026 17:12:13 +0000 Subject: [PATCH 1/2] Fix render issue for tools that are streamed in parts. --- .../console/observers/tool_call_display.py | 105 +++++++++++++++++- 1 file changed, 100 insertions(+), 5 deletions(-) diff --git a/python/samples/02-agents/harness/console/observers/tool_call_display.py b/python/samples/02-agents/harness/console/observers/tool_call_display.py index e9999a9db6..d270451947 100644 --- a/python/samples/02-agents/harness/console/observers/tool_call_display.py +++ b/python/samples/02-agents/harness/console/observers/tool_call_display.py @@ -4,6 +4,7 @@ from __future__ import annotations +import json from typing import TYPE_CHECKING, Any from ..formatters import build_default_formatters, format_tool_call @@ -12,6 +13,7 @@ if TYPE_CHECKING: from agent_framework import Agent, Content + from ..app_state import FollowUpAction from ..formatters import ToolCallFormatter from ..state_driver import IUXStateDriver @@ -21,6 +23,16 @@ class ToolCallDisplayObserver(ConsoleObserver): Shows tool calls with a 🔧 prefix and uses the formatter system to display them in a user-friendly format. + + Streaming clients (e.g. the OpenAI/Foundry Responses API) emit a separate + ``function_call`` content item for every ``arguments`` delta — each sharing + the same ``call_id`` and ``name`` but carrying only a partial fragment of the + JSON arguments. Printing one line per content item therefore repeats a single + tool call many times (scaling with argument size). To avoid that, this + observer buffers the argument fragments per ``call_id`` and emits exactly one + line once the accumulated arguments are complete (i.e. parse as valid JSON, + or arrive already-coalesced as a mapping). Any call that never reaches a + complete state is flushed when streaming completes. """ def __init__(self, formatters: list[ToolCallFormatter] | None = None) -> None: @@ -31,6 +43,10 @@ def __init__(self, formatters: list[ToolCallFormatter] | None = None) -> None: default formatters from build_default_formatters(). """ self._formatters = formatters or build_default_formatters() + # call_id -> {"name": str, "arguments": str | dict} + self._pending: dict[str, dict[str, Any]] = {} + # call_ids already displayed in the current stream (avoid duplicates). + self._displayed: set[str] = set() async def on_content( self, @@ -39,7 +55,7 @@ async def on_content( agent: Agent, session: Any, ) -> None: - """Display function call content. + """Buffer streamed function-call fragments and display each call once. Args: ux: The UX state driver for UI updates. @@ -47,7 +63,86 @@ async def on_content( agent: The AI agent. session: The agent session. """ - # Check if this is a function call content type - if content.type == "function_call": - formatted = format_tool_call(self._formatters, content) - ux.append_info_line(f"🔧 {formatted}", "yellow") + if content.type != "function_call": + return + + # Group streamed fragments by call_id. Some providers may omit a call_id; + # fall back to a name-derived key so distinct unnamed calls don't merge. + call_id = content.call_id or f"__noid_{content.name or 'unknown'}" + + if call_id in self._displayed: + return + + entry = self._pending.setdefault(call_id, {"name": content.name, "arguments": ""}) + if content.name and not entry["name"]: + entry["name"] = content.name + + args = content.arguments + if isinstance(args, str): + # Streaming delta fragment — concatenate. + entry["arguments"] = (entry["arguments"] or "") + args + elif args is not None: + # Already-coalesced arguments (e.g. a mapping) — use directly. + entry["arguments"] = args + + if self._is_complete(entry["arguments"]): + self._flush(ux, call_id) + + async def on_stream_complete( + self, + ux: IUXStateDriver, + agent: Agent, + session: Any, + ) -> list[FollowUpAction] | None: + """Flush buffered calls that never reached a complete state, then reset. + + Args: + ux: The UX state driver for UI updates. + agent: The AI agent. + session: The agent session. + + Returns: + Always None; this observer produces no follow-up actions. + """ + for call_id in list(self._pending): + self._flush(ux, call_id) + self._pending.clear() + self._displayed.clear() + return None + + @staticmethod + def _is_complete(arguments: Any) -> bool: + """Return True when the accumulated arguments form a complete payload. + + A mapping is already complete. A string is complete once it parses as + JSON (partial fragments of a streamed JSON object will not parse until + the closing brace arrives; a no-argument call streams ``"{}"`` which + parses immediately). + """ + if isinstance(arguments, str): + if not arguments: + return False + try: + json.loads(arguments) + except (json.JSONDecodeError, TypeError): + return False + return True + # Non-string (mapping / None handled by caller) is treated as complete. + return arguments is not None + + def _flush(self, ux: IUXStateDriver, call_id: str) -> None: + """Format and display a buffered call exactly once.""" + entry = self._pending.pop(call_id, None) + if entry is None or call_id in self._displayed: + return + self._displayed.add(call_id) + + from agent_framework import Content + + call = Content.from_function_call( + call_id=call_id, + name=entry["name"] or "Unknown", + arguments=entry["arguments"] or None, + ) + formatted = format_tool_call(self._formatters, call) + ux.append_info_line(f"🔧 {formatted}", "yellow") From e224153c7b3d4170f8e00a11e10b9d08b8d9b642 Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Tue, 16 Jun 2026 17:35:04 +0000 Subject: [PATCH 2/2] Address PR review: missing call_id fallback, empty-mapping args, _is_complete perf - Print call_id-less function calls as-is instead of merging under a name-derived key (which could drop distinct unnamed calls). - Preserve an empty {} mapping rather than coercing it to None. - Add a structural bracket-balance gate before json.loads in _is_complete to avoid O(n^2) re-parsing of growing streamed arguments. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../console/observers/tool_call_display.py | 34 +++++++++++++++---- 1 file changed, 28 insertions(+), 6 deletions(-) diff --git a/python/samples/02-agents/harness/console/observers/tool_call_display.py b/python/samples/02-agents/harness/console/observers/tool_call_display.py index d270451947..97c5cacece 100644 --- a/python/samples/02-agents/harness/console/observers/tool_call_display.py +++ b/python/samples/02-agents/harness/console/observers/tool_call_display.py @@ -66,9 +66,14 @@ async def on_content( if content.type != "function_call": return - # Group streamed fragments by call_id. Some providers may omit a call_id; - # fall back to a name-derived key so distinct unnamed calls don't merge. - call_id = content.call_id or f"__noid_{content.name or 'unknown'}" + # Streamed fragments are coalesced by call_id. If a provider omits the + # call_id, fragments cannot be reliably grouped, so fall back to the + # original behavior — display the item as-is — rather than risk merging + # (and then dropping) distinct calls under a shared synthetic key. + call_id = content.call_id + if not call_id: + self._display(ux, content) + return if call_id in self._displayed: return @@ -120,10 +125,17 @@ def _is_complete(arguments: Any) -> bool: parses immediately). """ if isinstance(arguments, str): - if not arguments: + stripped = arguments.strip() + if not stripped: + return False + # Cheap structural gate: a complete JSON object/array opens and + # closes with matching brackets. This rejects growing partial + # fragments in O(1) so json.loads only runs on a plausibly-complete + # payload, avoiding O(n^2) re-parsing across many streamed deltas. + if not ((stripped[0] == "{" and stripped[-1] == "}") or (stripped[0] == "[" and stripped[-1] == "]")): return False try: - json.loads(arguments) + json.loads(stripped) except (json.JSONDecodeError, TypeError): return False return True @@ -139,10 +151,20 @@ def _flush(self, ux: IUXStateDriver, call_id: str) -> None: from agent_framework import Content + # Preserve an empty mapping ("{}") as-is; only treat an empty *string* + # (no arguments were ever streamed) as "no arguments". + arguments = entry["arguments"] + if arguments == "": + arguments = None + call = Content.from_function_call( call_id=call_id, name=entry["name"] or "Unknown", - arguments=entry["arguments"] or None, + arguments=arguments, ) + self._display(ux, call) + + def _display(self, ux: IUXStateDriver, call: Content) -> None: + """Format and write a single tool-call line.""" formatted = format_tool_call(self._formatters, call) ux.append_info_line(f"🔧 {formatted}", "yellow")