Agent Engine `:streamQuery?alt=sse` + AG-UI: yielding SSE strings gets JSON-encoded (double-encoding), breaking direct CopilotKit compatibility

[diefergil]

🎤 Tell us your idea

Agent Engine :streamQuery?alt=sse + AG-UI: yielding SSE strings gets JSON-encoded (double-encoding), breaking direct CopilotKit compatibility

Summary

I’m deploying an ADK agent to Vertex AI Agent Engine (using vertexai.agent_engines.templates.adk.AdkApp) and trying to expose it as an AG-UI-compatible streaming endpoint so that a frontend like CopilotKit can connect directly.

However, when my async_stream_query generator yields SSE-formatted strings (e.g. "data: {...}\n\n" produced by an AG-UI EventEncoder), Agent Engine appears to serialize those strings as JSON strings inside the SSE transport. This results in “SSE inside JSON inside SSE” (double-encoding) and breaks clients that expect standard AG-UI SSE.

Goal

Enable a clean, direct path:

• Agent Engine :streamQuery?alt=sse ➜ valid AG-UI SSE
  without requiring an external proxy/middleware.

Current behavior

If the generator yields a string like:

• data: {"type":"RUN_STARTED",...}\n\n

The client receives SSE where the payload is a quoted JSON string containing escaped newlines, e.g. (illustrative):

data: "data: {\"type\":\"RUN_STARTED\",\"threadId\":\"t1\",\"runId\":\"r1\"}\n\n"

So the consumer would need to:

parse SSE

JSON-decode the string

parse SSE again

Expected behavior (one of these would work)

Option A (recommended): Document/enable a pattern where the generator yields structured events/objects (dict) and Agent Engine’s SSE transport emits:

data: {"type":"RUN_STARTED","threadId":"t1","runId":"r1"}

Option B: Support a “raw SSE passthrough” mode so that if the generator yields an SSE-formatted string, it is streamed verbatim when alt=sse is used.

Minimal repro
Backend (simplified)

from vertexai.agent_engines.templates.adk import AdkApp
from ag_ui.encoder import EventEncoder
from ag_ui.core import RunStartedEvent

class MyAgentEngineApp(AdkApp):
    async def async_stream_query(self, *, user_id: str, message: str, **kwargs):
        encoder = EventEncoder()
        # encoder.encode(...) returns a string like: "data: {...}\n\n"
        yield encoder.encode(RunStartedEvent(thread_id="t1", run_id="r1"))

        # ... later
        yield encoder.encode({"type": "RUN_FINISHED"})

Call

curl \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse" \
  -d '{
    "class_method":"stream_query",
    "input": {
      "user_id":"u_123",
      "session_id":"s_123",
      "message":"hello"
    }
  }'

Why this matters

AG-UI clients (and CopilotKit integrations that rely on AG-UI) expect standard SSE where each data: line is a JSON object for the event. Double-encoding forces brittle client-side hacks or extra infrastructure.

Questions

What is the recommended way to expose AG-UI from an ADK agent deployed on Agent Engine?

Is there a supported way to stream raw SSE (passthrough) from :streamQuery?alt=sse?

If the correct approach is “yield dicts/objects”:

could docs include a concrete example showing how to translate ADK/agent events → AG-UI events without using EventEncoder.encode() when running on Agent Engine?

References

References

• Agent Engine streaming API (:streamQuery?alt=sse): https://docs.cloud.google.com/agent-builder/agent-engine/use/custom

• AdkApp template docs: https://docs.cloud.google.com/agent-builder/agent-engine/develop/adk

• AG-UI server/streaming expectations: https://docs.agentwire.io/quickstart/build

[ahatem93]

Hello @diefergil

Do you find the best way to expose AG-UI from an ADK agent deployed on Agent Engine?

[diefergil]

Hello @diefergil

Do you find the best way to expose AG-UI from an ADK agent deployed on Agent Engine?

Im using this workaround:

import asyncio
import json
import uuid
from collections.abc import AsyncIterable
from typing import Any

from ag_ui.core import RunErrorEvent, RunFinishedEvent, RunStartedEvent
from ag_ui.encoder import EventEncoder
from ag_ui_adk.event_translator import EventTranslator
from google.adk.events.event import Event
from vertexai.agent_engines.templates.adk import AdkApp

class AgUiAdkApp(AdkApp):
    """ADK Application that emits AG-UI compatible SSE events.

    Bridges Vertex AI ADK to AG-UI/CopilotKit protocol using
    ag_ui_adk.EventTranslator and ag_ui.encoder.EventEncoder.

    The EventEncoder ensures proper SSE formatting with:
    - Correct field aliases (messageId instead of message_id)
    - SSE data prefix format
    - Exclusion of None fields

    Note: Vertex AI Agent Engine will double-encode the SSE string as JSON.
    The frontend must parse the outer JSON string first, then process SSE.

    This class enables integration with:
    - CopilotKit frontends (useCoAgent, useCopilotChat)
    - Any AG-UI protocol compliant UI framework
    - Custom frontends using SSE streaming

    Inherits from BaseAgentEngineApp:
    - set_up(): Initializes vertexai, telemetry, and logging
    - register_feedback(): Validates and logs feedback
    - register_operations(): Registers feedback operation
    """

    def __init__(
        self,
        *args: Any,
        verbose_logging: bool = False,
        **kwargs: Any,
    ) -> None:
        """Initialize with EventTranslator and EventEncoder.

        Args:
            verbose_logging: Enable detailed Cloud Logging
            *args: Additional positional arguments for AdkApp
            **kwargs: Additional keyword arguments for AdkApp
        """
        super().__init__(*args, **kwargs)
        self.translator = EventTranslator()
        self.encoder = EventEncoder()  # Official AG-UI encoder for proper SSE format
        self.verbose_logging = verbose_logging

    async def async_stream_query(
        self,
        *,
        message: str | dict[str, Any],
        user_id: str,
        session_id: str | None = None,
        run_config: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[str]:
        """Streams AG-UI compatible SSE responses.

        Overrides the parent method to emit SSE-formatted events that are
        compatible with AG-UI protocol and CopilotKit's HttpAgent.

        Uses:
        - ag_ui_adk.EventTranslator for ADK to AG-UI event translation
        - ag_ui.encoder.EventEncoder for proper SSE formatting with aliases

        Note: The yielded strings are SSE-formatted (e.g., "data: {...}\\n\\n").
        Vertex AI Agent Engine will serialize these as JSON strings, resulting
        in double-encoding that the frontend must handle.

        Args:
            message: The message to stream responses for (string or Content dict)
            user_id: The ID of the user
            session_id: Optional session ID. If not provided, creates new session
            run_config: Optional run configuration
            **kwargs: Additional keyword arguments

        Yields:
            SSE-formatted strings with properly aliased AG-UI events
        """
        # Generate thread_id and run_id for AG-UI protocol
        # Note: AG-UI uses 'thread_id' for what ADK calls 'session_id'
        thread_id = session_id or str(uuid.uuid4())
        run_id = str(uuid.uuid4())
        cloud_logger = _get_cloud_logger()

        # Call parent's async_stream_query to get ADK event dicts
        adk_event_count = 0
        ag_ui_event_count = 0

        cookies_dict = kwargs.pop("cookies", {})
        context = RequestContext(cookies=Cookies.model_validate(cookies_dict))
        token = request_context.set(context)

        try:
            # Emit RUN_STARTED event at the beginning (required by AG-UI protocol)
            run_started_event = RunStartedEvent(
                thread_id=thread_id,
                run_id=run_id,
            )
            ag_ui_event_count += 1
            if cloud_logger:
                cloud_logger.log_text(
                    f"AG-UI: Emitting RUN_STARTED (thread={thread_id}, run={run_id})",
                    severity="INFO",
                )
            yield self.encoder.encode(run_started_event)

            async for adk_event_dict in super().async_stream_query(
                message=message,
                user_id=user_id,
                session_id=session_id,
                run_config=run_config,
                **kwargs,
            ):
                adk_event_count += 1

                # Convert dict to Event object for EventTranslator
                adk_event = Event.model_validate(adk_event_dict)

                # Verbose logging (optional)
                if self.verbose_logging and cloud_logger:
                    keys = list(adk_event_dict.keys())
                    cloud_logger.log_text(
                        f"ADK #{adk_event_count}: {keys}", severity="INFO"
                    )

                # Translate ADK event to AG-UI events using the library
                async for ag_ui_event in self.translator.translate(
                    adk_event, thread_id, run_id
                ):
                    ag_ui_event_count += 1

                    event_msg = f"AG-UI #{ag_ui_event_count}: {ag_ui_event.type}"
                    if cloud_logger:
                        cloud_logger.log_text(event_msg, severity="INFO")
                        _safe_log_struct(ag_ui_event.model_dump(), severity="INFO")

                    # Use official EventEncoder for proper SSE format with aliases
                    yield self.encoder.encode(ag_ui_event)

            # Emit RUN_FINISHED on successful completion
            run_finished_event = RunFinishedEvent(
                thread_id=thread_id,
                run_id=run_id,
            )
            ag_ui_event_count += 1
            if cloud_logger:
                cloud_logger.log_text(
                    f"AG-UI: Emitting RUN_FINISHED (thread={thread_id}, run={run_id})",
                    severity="INFO",
                )
            yield self.encoder.encode(run_finished_event)

        except Exception as e:
            # Emit RUN_ERROR on failure
            if cloud_logger:
                cloud_logger.log_text(
                    f"AG-UI: Emitting RUN_ERROR: {e!s}",
                    severity="ERROR",
                )
            error_event = RunErrorEvent(
                message=str(e),
                code="EXECUTION_ERROR",
            )
            yield self.encoder.encode(error_event)

        finally:
            summary_msg = f"Complete - ADK:{adk_event_count}, AG-UI:{ag_ui_event_count}"
            if cloud_logger:
                cloud_logger.log_text(summary_msg, severity="INFO")

            request_context.reset(token)

            # Small delay to ensure final events are flushed before exit
            await asyncio.sleep(0.1)

[llalitkumarrr]

Hello @diefergil and @ahatem93 ,

Has the available fix resolved your issue or do you require further support? If you still need assistance, please upgrade to the latest Google ADK version (1.23.0) and let us know the results so we can help you better.