Add web search grounding feature

Implement Google Custom Search API integration for real-time web
grounding. Tools can now return dict data directly to the model
instead of client-bound frames.

Author: oadultradeepfield

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "pydantic>=2.10.0",
     "pydantic-settings>=2.6.0",
     "python-dotenv>=1.0.0",
+    "httpx>=0.27.0",
 ]
 
 [project.optional-dependencies]

src/actions/base.py

@@ -125,5 +125,5 @@ async def tool_fn(
         self,
         ctx: "ConnectionContext",
         **kwargs: Any,
-    ) -> OutboundFrame:
+    ) -> OutboundFrame | dict[str, Any]:
         pass

src/actions/server/web_search.py

@@ -0,0 +1,81 @@
+"""Search the web for up-to-date information using Google Custom Search API."""
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+from src.actions.base import ServerEventEmitter, register_server_event
+from src.schema.frames import OutboundFrame
+
+if TYPE_CHECKING:
+    from src.engine.connection_context import ConnectionContext
+
+logger = logging.getLogger(__name__)
+
+
+@register_server_event("WEB_SEARCH")
+class WebSearchEmitter(ServerEventEmitter):
+    """Search the web for up-to-date information."""
+
+    event_type = "WEB_SEARCH"
+    description = (
+        "Search the web for up-to-date information. Use proactively when "
+        "the user asks about recent events, weather, prices, or time-sensitive topics."
+    )
+    parameters = {
+        "query": {"type": "string", "description": "The search query"},
+        "num_results": {
+            "type": "integer",
+            "description": "Number of results (default: 5, max: 10)",
+        },
+    }
+
+    async def tool_fn(  # type: ignore[override]
+        self, _ctx: "ConnectionContext", **kwargs: Any
+    ) -> OutboundFrame | dict[str, Any]:
+        from src.config import get_settings
+
+        settings = get_settings()
+        api_key = settings.google_search_api_key
+        engine_id = settings.google_search_engine_id
+
+        if not api_key or not engine_id:
+            logger.warning("Google Search API credentials not configured")
+            return {"error": "Search not configured", "results": []}
+
+        query = kwargs.get("query", "")
+        if not query:
+            return {"error": "No search query provided", "results": []}
+
+        num_results = min(kwargs.get("num_results", 5), 10)
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.get(
+                    "https://www.googleapis.com/customsearch/v1",
+                    params={
+                        "key": api_key,
+                        "cx": engine_id,
+                        "q": query,
+                        "num": num_results,
+                    },
+                    timeout=10.0,
+                )
+                response.raise_for_status()
+                data = response.json()
+            except httpx.HTTPError as e:
+                logger.exception(f"Search API error: {e}")
+                return {"error": f"Search failed: {e}", "results": []}
+
+        items = data.get("items", [])
+        results = [
+            {
+                "title": item.get("title", ""),
+                "link": item.get("link", ""),
+                "snippet": item.get("snippet", ""),
+            }
+            for item in items
+        ]
+
+        return {"query": query, "results": results}

src/agent/runner.py

@@ -268,7 +268,7 @@ async def _handle_function_call(
             function_call: The FunctionCall to process
 
         Returns:
-            OutboundFrame from the tool execution
+            OutboundFrame from the tool execution, or None if result sent to model
         """
         from src.actions.base import get_server_event_emitter
 
@@ -284,8 +284,19 @@ async def _handle_function_call(
             return None
 
         try:
-            frame = await emitter.tool_fn(self.ctx, **args)
-            return frame
+            result = await emitter.tool_fn(self.ctx, **args)
+
+            if isinstance(result, dict):
+                response = types.FunctionResponse(name=tool_name, response=result)
+                content = types.Content(
+                    role="user",
+                    parts=[types.Part(function_response=response)],
+                )
+                if self.adk_queue:
+                    self.adk_queue.send_content(content)
+                return None
+
+            return result
         except (TypeError, ValueError, KeyError) as e:
             logger.exception(f"Error executing tool {tool_name}: {e}")
             return OutboundFrame(

src/config.py

@@ -14,6 +14,8 @@ class Settings(BaseSettings):
     session_grace_minutes: int = 10
     app_name: str = "jemmie"
     log_level: str = "INFO"
+    google_search_api_key: str | None = None
+    google_search_engine_id: str | None = None
 
 
 @lru_cache

src/prompts/jemmie.py

@@ -38,6 +38,10 @@
 7. **Ending Call (END_CALL)**:
    - When the user wants to verbally end the interaction (e.g. "I gotta go", "hang up", "bye bye"), gracefully say goodbye and immediately invoke `end_call` to automatically hang up for them.
 
+8. **Web Search (WEB_SEARCH)**:
+   - When asked about current events, weather, sports, stock prices, or anything time-sensitive, proactively use `web_search` to get accurate, up-to-date information.
+   - After searching, naturally incorporate findings: "According to recent reports..."
+
 ## Personality
 
 - Warm, approachable, genuinely helpful, and incredibly empathetic.

tests/test_actions.py

@@ -192,3 +192,66 @@ def test_get_server_event_emitter_case_insensitive() -> None:
 def test_get_server_event_emitter_not_found() -> None:
     emitter = get_server_event_emitter("nonexistent_tool")
     assert emitter is None
+
+
+def test_web_search_registered() -> None:
+    emitter = get_server_event_emitter("web_search")
+    assert emitter is not None
+    assert emitter.event_type == "WEB_SEARCH"
+
+
+def test_web_search_tool_definition() -> None:
+    tools = get_all_server_tools()
+    web_search_tool = next((t for t in tools if t["name"] == "web_search"), None)
+    assert web_search_tool is not None
+    assert "query" in web_search_tool["parameters"]["properties"]
+    assert "num_results" in web_search_tool["parameters"]["properties"]
+
+
+@pytest.mark.asyncio
+async def test_web_search_no_credentials(mock_websocket: MockWebSocket) -> None:
+    from src.actions.server.web_search import WebSearchEmitter
+
+    emitter = WebSearchEmitter()
+    ctx = ConnectionContext(
+        device_id="test-device",
+        session_id="test-session",
+        websocket=mock_websocket,
+    )
+
+    result = await emitter.tool_fn(ctx, query="test query")
+    assert isinstance(result, dict)
+    assert result["error"] == "Search not configured"
+    assert result["results"] == []
+
+
+@pytest.mark.asyncio
+async def test_web_search_empty_query(
+    mock_websocket: MockWebSocket, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from src.actions.server.web_search import WebSearchEmitter
+    from src.config import get_settings
+
+    monkeypatch.setenv("GOOGLE_SEARCH_API_KEY", "test-key")
+    monkeypatch.setenv("GOOGLE_SEARCH_ENGINE_ID", "test-id")
+    get_settings.cache_clear()
+
+    emitter = WebSearchEmitter()
+    ctx = ConnectionContext(
+        device_id="test-device",
+        session_id="test-session",
+        websocket=mock_websocket,
+    )
+
+    result = await emitter.tool_fn(ctx, query="")
+    assert isinstance(result, dict)
+    assert result["error"] == "No search query provided"
+    assert result["results"] == []
+
+
+def test_web_search_returns_dict_not_frame() -> None:
+    """Verify web_search returns a dict (model-bound), not OutboundFrame (client-bound)."""
+    emitter = get_server_event_emitter("web_search")
+    assert emitter is not None
+    tool_def = emitter.get_tool_definition()
+    assert tool_def["name"] == "web_search"