refactor(demo): move native event away from devops agent def

demo/agent-server/src/agent_server/agents/devops.py

@@ -15,6 +15,7 @@
 
 from .. import protocol
 from . import devops_agent
+from .google_adk import to_native_event
 
 logger = logging.getLogger("agent_server.devops")
 
@@ -59,7 +60,7 @@ async def run(
 
         try:
             async for adk_event in events:
-                native_event = devops_agent.to_native_event(adk_event)
+                native_event = to_native_event(adk_event)
                 if native_event is not None:
                     yield native_event
         except Exception as exception:  # noqa: BLE001 — degrade to a visible error event

demo/agent-server/src/agent_server/agents/devops_agent.py

@@ -3,8 +3,8 @@
 The tools + ``agent``, and how to invoke it: ``stream`` (plain ADK runner) and
 ``stream_as`` (the same agent gated by HexGate policy). Vendored from
 ``hexgate/examples/devops_agent.py``. The HexaUI contract wrapper that the
-server runs lives in ``devops.py``; ADK ``Event`` → native projection is
-``to_native_event`` below.
+server runs lives in ``devops.py``; the ADK ``Event`` → native projection lives
+in ``google_adk``.
 
 One agent definition; the caller's ``role`` (viewer < operator < admin) is what
 flips the decision — the policy gates ``scale_deployment`` on the replica count
@@ -138,60 +138,3 @@ async def stream_as(text: str, *, user_id: str, role: str) -> AsyncIterator[Any]
     runner = HexgateRunner(agent=agent, app_name=_APP_NAME, session_service=session_service)
     async for event in runner.run_async(new_message=_message(text), user=user):
         yield event
-
-
-# ── ADK Event → HexaUI native event ──────────────────────────────────────────
-
-
-def to_native_event(event: Any) -> dict | None:
-    """Project one ADK ``Event`` into the native JSON the proxy's
-    ``GoogleADKTranslator`` reads (``None`` to drop it).
-
-    Mirrors the wire shape the translator expects: ``author`` + ``content.parts``
-    of ``text`` / ``function_call`` / ``function_response``, with ``partial`` and
-    ``turn_complete`` carried through so block framing stays correct.
-    """
-    content = getattr(event, "content", None)
-    raw_parts = getattr(content, "parts", None) or []
-
-    parts: list[dict] = []
-    for part in raw_parts:
-        func_call = getattr(part, "function_call", None)
-        func_resp = getattr(part, "function_response", None)
-        text = getattr(part, "text", None)
-        if func_call is not None:
-            parts.append(
-                {
-                    "function_call": {
-                        "id": getattr(func_call, "id", None) or "",
-                        "name": getattr(func_call, "name", "tool") or "tool",
-                        "args": dict(getattr(func_call, "args", None) or {}),
-                    }
-                }
-            )
-        elif func_resp is not None:
-            parts.append(
-                {
-                    "function_response": {
-                        "id": getattr(func_resp, "id", None) or "",
-                        "name": getattr(func_resp, "name", "tool") or "tool",
-                        "response": getattr(func_resp, "response", None),
-                    }
-                }
-            )
-        elif text:
-            parts.append({"text": text})
-
-    turn_complete = bool(getattr(event, "turn_complete", False))
-    if not parts and not turn_complete:
-        return None
-
-    native: dict[str, Any] = {
-        "author": getattr(event, "author", None) or "assistant",
-        "content": {"parts": parts},
-    }
-    if getattr(event, "partial", False):
-        native["partial"] = True
-    if turn_complete:
-        native["turn_complete"] = True
-    return native

demo/agent-server/src/agent_server/agents/google_adk.py

@@ -0,0 +1,64 @@
+"""Project Google ADK runtime events into HexaUI native events.
+
+Reusable for any real ADK runtime: ``to_native_event`` maps one ADK ``Event``
+to the native JSON event the proxy's ``GoogleADKTranslator`` reads. The
+agent-server side counterpart of ``openai_agents`` / ``langchain_events``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def to_native_event(event: Any) -> dict | None:
+    """Project one ADK ``Event`` into the native JSON the proxy's
+    ``GoogleADKTranslator`` reads (``None`` to drop it).
+
+    Mirrors the wire shape the translator expects: ``author`` + ``content.parts``
+    of ``text`` / ``function_call`` / ``function_response``, with ``partial`` and
+    ``turn_complete`` carried through so block framing stays correct.
+    """
+    content = getattr(event, "content", None)
+    raw_parts = getattr(content, "parts", None) or []
+
+    parts: list[dict] = []
+    for part in raw_parts:
+        func_call = getattr(part, "function_call", None)
+        func_resp = getattr(part, "function_response", None)
+        text = getattr(part, "text", None)
+        if func_call is not None:
+            parts.append(
+                {
+                    "function_call": {
+                        "id": getattr(func_call, "id", None) or "",
+                        "name": getattr(func_call, "name", "tool") or "tool",
+                        "args": dict(getattr(func_call, "args", None) or {}),
+                    }
+                }
+            )
+        elif func_resp is not None:
+            parts.append(
+                {
+                    "function_response": {
+                        "id": getattr(func_resp, "id", None) or "",
+                        "name": getattr(func_resp, "name", "tool") or "tool",
+                        "response": getattr(func_resp, "response", None),
+                    }
+                }
+            )
+        elif text:
+            parts.append({"text": text})
+
+    turn_complete = bool(getattr(event, "turn_complete", False))
+    if not parts and not turn_complete:
+        return None
+
+    native: dict[str, Any] = {
+        "author": getattr(event, "author", None) or "assistant",
+        "content": {"parts": parts},
+    }
+    if getattr(event, "partial", False):
+        native["partial"] = True
+    if turn_complete:
+        native["turn_complete"] = True
+    return native