From a01493d762efbda0144e0f28e7cc83f5ea50acf0 Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Wed, 17 Jun 2026 16:20:09 +0000 Subject: [PATCH 1/2] Adding an observer to the python harness for web search tools --- .../harness/console/observers/__init__.py | 4 + .../console/observers/web_search_display.py | 143 ++++++++++++++++++ 2 files changed, 147 insertions(+) create mode 100644 python/samples/02-agents/harness/console/observers/web_search_display.py diff --git a/python/samples/02-agents/harness/console/observers/__init__.py b/python/samples/02-agents/harness/console/observers/__init__.py index 7200939d74..0642e2c544 100644 --- a/python/samples/02-agents/harness/console/observers/__init__.py +++ b/python/samples/02-agents/harness/console/observers/__init__.py @@ -19,6 +19,7 @@ from .tool_approval import ToolApprovalObserver from .tool_call_display import ToolCallDisplayObserver from .usage_display import UsageDisplayObserver +from .web_search_display import WebSearchDisplayObserver if TYPE_CHECKING: from agent_framework import Agent @@ -45,6 +46,7 @@ def build_default_observers() -> list[ConsoleObserver]: return [ TextOutputObserver(), ToolCallDisplayObserver(), + WebSearchDisplayObserver(), ErrorDisplayObserver(), UsageDisplayObserver(), ReasoningDisplayObserver(), @@ -95,6 +97,7 @@ def build_observers_with_planning( return [ ToolCallDisplayObserver(), + WebSearchDisplayObserver(), ToolApprovalObserver(), ErrorDisplayObserver(), ReasoningDisplayObserver(), @@ -117,6 +120,7 @@ def build_observers_with_planning( "ToolApprovalObserver", "ToolCallDisplayObserver", "UsageDisplayObserver", + "WebSearchDisplayObserver", "build_default_observers", "build_observers_with_planning", ] diff --git a/python/samples/02-agents/harness/console/observers/web_search_display.py b/python/samples/02-agents/harness/console/observers/web_search_display.py new file mode 100644 index 0000000000..4fcab660d3 --- /dev/null +++ b/python/samples/02-agents/harness/console/observers/web_search_display.py @@ -0,0 +1,143 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Web search display observer for showing search activity in the console. + +Displays web search activity as it streams in from the API, showing search +queries, page opens, and find-in-page actions with 🌐 prefix. + +The actual details (queries, URLs, sources) come from the ``search_tool_result`` +content emitted when the search completes (``response.output_item.done``). +The initial ``search_tool_call`` is emitted when the item is first added and +typically has an empty or incomplete action. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from .base import ConsoleObserver + +if TYPE_CHECKING: + from agent_framework import Agent, Content + + from ..state_driver import IUXStateDriver + +_MAX_QUERY_DISPLAY_LENGTH = 120 + + +class WebSearchDisplayObserver(ConsoleObserver): + """Displays web search activity in the scroll area. + + Shows search queries, page opens, and find-in-page actions. Details are + extracted from ``search_tool_result`` content (the completed action), which + contains the full action type, queries, URLs, and sources. + """ + + async def on_content( + self, + ux: IUXStateDriver, + content: Content, + agent: Agent, + session: Any, + ) -> None: + """Display web search activity from search content items. + + Args: + ux: The UX state driver for UI updates. + content: The content item to check for search activity. + agent: The AI agent. + session: The agent session. + """ + if content.type == "search_tool_result": + self._display_search_result(ux, content) + + def _display_search_result(self, ux: IUXStateDriver, content: Content) -> None: + """Display a completed search tool result with action details.""" + tool_name = getattr(content, "tool_name", None) or "web_search" + if tool_name != "web_search": + return + + result = getattr(content, "result", None) + if not isinstance(result, dict): + ux.append_info_line("🌐 Web Search", "cyan") + return + + action = result.get("action") + if not isinstance(action, dict): + ux.append_info_line("🌐 Web Search", "cyan") + return + + action_type = action.get("type") + + if action_type == "search": + self._display_search_action(ux, action) + elif action_type == "open_page": + self._display_open_page_action(ux, action) + elif action_type == "find_in_page": + self._display_find_in_page_action(ux, action) + else: + ux.append_info_line("🌐 Web Search", "cyan") + + def _display_search_action(self, ux: IUXStateDriver, action: dict) -> None: + """Display a search action with queries and optional sources.""" + queries = action.get("queries") or [] + if not queries: + # Fall back to the single "query" field + query = action.get("query") + if query: + queries = [query] + + if not queries: + ux.append_info_line("🌐 Web Search: search", "cyan") + return + + sources = action.get("sources") or [] + has_sources = len(sources) > 0 + + lines = ["🌐 Web Search: search"] + for i, query in enumerate(queries): + connector = "├─" if (i < len(queries) - 1 or has_sources) else "└─" + query_text = _truncate(str(query), _MAX_QUERY_DISPLAY_LENGTH) + lines.append(f'\n {connector} "{query_text}"') + + if has_sources: + lines.append("\n │") + for i, source in enumerate(sources): + connector = "├─" if i < len(sources) - 1 else "└─" + line = _format_source(source) + lines.append(f"\n {connector} {line}") + + ux.append_info_line("".join(lines), "cyan") + + def _display_open_page_action(self, ux: IUXStateDriver, action: dict) -> None: + """Display an open page action.""" + url = action.get("url") or "(unknown)" + ux.append_info_line( + f"🌐 Web Search: open page\n └─ {url}", + "cyan", + ) + + def _display_find_in_page_action(self, ux: IUXStateDriver, action: dict) -> None: + """Display a find-in-page action.""" + url = action.get("url") or "(unknown)" + pattern = action.get("pattern") or "(unknown)" + ux.append_info_line( + f'🌐 Web Search: find in page\n ├─ "{_truncate(pattern, _MAX_QUERY_DISPLAY_LENGTH)}"\n └─ {url}', + "cyan", + ) + + +def _truncate(text: str, max_length: int) -> str: + """Truncate text to max length with ellipsis.""" + return text if len(text) <= max_length else text[: max_length - 1] + "…" + + +def _format_source(source: Any) -> str: + """Format a source entry for display.""" + if isinstance(source, dict): + url = source.get("url") or source.get("uri") or "(unknown)" + title = source.get("title") + if title: + return f"{_truncate(title, _MAX_QUERY_DISPLAY_LENGTH)} — {url}" + return str(url) + return str(source) From 65a03ef3a951ebf195b6c53ace872a3985ad40c4 Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Wed, 17 Jun 2026 16:29:49 +0000 Subject: [PATCH 2/2] Escape dynamic strings with rich.markup.escape() in WebSearchDisplayObserver Apply rich.markup.escape() to all user/tool-provided strings (queries, URLs, titles, patterns) before interpolation into Rich-markup-enabled output. This prevents characters like '['/']' from being interpreted as Rich markup tags. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../console/observers/web_search_display.py | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/python/samples/02-agents/harness/console/observers/web_search_display.py b/python/samples/02-agents/harness/console/observers/web_search_display.py index 4fcab660d3..190cec1236 100644 --- a/python/samples/02-agents/harness/console/observers/web_search_display.py +++ b/python/samples/02-agents/harness/console/observers/web_search_display.py @@ -15,6 +15,8 @@ from typing import TYPE_CHECKING, Any +from rich.markup import escape + from .base import ConsoleObserver if TYPE_CHECKING: @@ -97,7 +99,7 @@ def _display_search_action(self, ux: IUXStateDriver, action: dict) -> None: lines = ["🌐 Web Search: search"] for i, query in enumerate(queries): connector = "├─" if (i < len(queries) - 1 or has_sources) else "└─" - query_text = _truncate(str(query), _MAX_QUERY_DISPLAY_LENGTH) + query_text = escape(_truncate(str(query), _MAX_QUERY_DISPLAY_LENGTH)) lines.append(f'\n {connector} "{query_text}"') if has_sources: @@ -111,7 +113,7 @@ def _display_search_action(self, ux: IUXStateDriver, action: dict) -> None: def _display_open_page_action(self, ux: IUXStateDriver, action: dict) -> None: """Display an open page action.""" - url = action.get("url") or "(unknown)" + url = escape(str(action.get("url") or "(unknown)")) ux.append_info_line( f"🌐 Web Search: open page\n └─ {url}", "cyan", @@ -119,10 +121,10 @@ def _display_open_page_action(self, ux: IUXStateDriver, action: dict) -> None: def _display_find_in_page_action(self, ux: IUXStateDriver, action: dict) -> None: """Display a find-in-page action.""" - url = action.get("url") or "(unknown)" - pattern = action.get("pattern") or "(unknown)" + url = escape(str(action.get("url") or "(unknown)")) + pattern = escape(_truncate(str(action.get("pattern") or "(unknown)"), _MAX_QUERY_DISPLAY_LENGTH)) ux.append_info_line( - f'🌐 Web Search: find in page\n ├─ "{_truncate(pattern, _MAX_QUERY_DISPLAY_LENGTH)}"\n └─ {url}', + f'🌐 Web Search: find in page\n ├─ "{pattern}"\n └─ {url}', "cyan", ) @@ -135,9 +137,9 @@ def _truncate(text: str, max_length: int) -> str: def _format_source(source: Any) -> str: """Format a source entry for display.""" if isinstance(source, dict): - url = source.get("url") or source.get("uri") or "(unknown)" + url = escape(str(source.get("url") or source.get("uri") or "(unknown)")) title = source.get("title") if title: - return f"{_truncate(title, _MAX_QUERY_DISPLAY_LENGTH)} — {url}" - return str(url) - return str(source) + return f"{escape(_truncate(str(title), _MAX_QUERY_DISPLAY_LENGTH))} — {url}" + return url + return escape(str(source))