No first-class way to serve the ADK data plane (`streamQuery`/sessions) **and** A2A 1.0 on a single Agent Engine

[szamcsi-at-google]

🔴 Required Information

Describe the Bug:

AdkApp and A2aAgent (the vertexai.agent_engines / google-cloud-aiplatform
templates) are mutually exclusive on a single Agent Engine, so there is no
supported way to deploy one reasoning engine that serves both the ADK data plane
and A2A:

• AdkApp.register_operations() returns only the ADK data-plane modes
  ("", async, stream, async_stream → stream_query, sessions, artifacts, …).
• A2aAgent.register_operations() returns only a2a_extension
  (on_message_send, on_get_task, …).

An agent that needs its ADK data plane (e.g. it relies on :streamQuery, or on
the durable self-:asyncQuery query-job pattern) and wants to be reachable
by external A2A clients must hand-build a custom hybrid template. We built and
verified one, but this should be first-class.

This is distinct from two existing issues: #6274 (upgrade a2a-sdk to 1.x)
fixes the version conflict but does not provide a hybrid deployment; #6220
(adk deploy doesn't provision :asyncQuery) is orthogonal. Neither removes the
need for the workaround here. (Treating this as a bug rather than a feature: the
platform already accepts a hybrid spec — see Observed Behavior — there is just no
template/option that produces one.)

Steps to Reproduce:

Mutual exclusivity is visible with no deploy and no credentials — just inspect
what each template registers:

1. Install google-adk and google-cloud-aiplatform (+ a2a-sdk 1.x).
2. Run the snippet under "Minimal Reproduction Code" below.
3. Observe that AdkApp registers only the ADK modes and A2aAgent registers
   only a2a_extension — with no option/flag/helper to get both on one engine.

Expected Behavior:

A first-class way to expose A2A alongside the ADK data plane on one Agent
Engine — e.g. an AdkApp(..., a2a=True, agent_card=...) option, or a supported
composition helper / template — so one engine exposes both :streamQuery
(+ sessions, + the durable self-:asyncQuery) and /a2a.

Observed Behavior:

Each template exposes only its own half:

AdkApp.register_operations()  -> ['', 'async', 'stream', 'async_stream']   # no A2A
A2aAgent.register_operations()-> ['a2a_extension']                          # no data plane

The platform itself does accept a hybrid: our hand-built subclass (an AdkApp
that composes an A2aAgent around the same ADK Runner and merges
register_operations()), deployed via the GenAI client, produced a single engine
whose spec has both surfaces, and both :streamQuery and /a2a (message:send)
respond on it:

class methods: {'': 4, 'async': 13, 'stream': 1, 'async_stream': 2, 'a2a_extension': 9}
top-level spec.agent_card: present

So the gap is purely the missing template/option, not a backend limitation.

Environment Details:

• ADK Library Version (pip show google-adk): 2.2.0 (2.3.0 still pins a2a-sdk>=0.3.4,<0.4)
• Desktop OS: Linux
• Python Version (python -V): 3.13

Model Information:

• Are you using LiteLLM: No
• Which model is being used: gemini-2.5-flash (irrelevant — this is a deploy/templating issue)

---

🟡 Optional Information

Additional Context:

Use case. An ADK orchestrator on Agent Engine that (1) runs long-running work
as durable query jobs it submits against itself (run_query_job / :asyncQuery,
which needs the ADK stream_query op registered on the engine) and (2) must be
reachable from a UI over the A2A protocol. A pure A2aAgent satisfies (2) but
drops the ADK ops, breaking (1); a pure AdkApp satisfies (1) but has no A2A.

What we had to build. Subclass AdkApp, compose an A2aAgent around the same
Runner, and merge the op maps:

class A2aAdkApp(AdkApp):
    def register_operations(self):
        ops = dict(super().register_operations())          # ADK data plane
        ops["a2a_extension"] = [                            # + A2A surface
            "on_message_send", "on_get_task", "on_list_tasks", "on_cancel_task",
            "on_create_task_push_notification_config",
            "on_get_task_push_notification_config",
            "on_list_task_push_notification_configs",
            "on_delete_task_push_notification_config",
            "on_get_extended_agent_card",
        ]
        return ops

    def set_up(self):
        super().set_up()                                   # builds the ADK Runner
        a2a = A2aAgent(agent_card=self.agent_card,
                       agent_executor_builder=lambda: <executor wrapping the Runner>)
        a2a.set_up()
        # delegate the on_* methods + rest_routes + agent_card to `a2a`

Deployed via the GenAI client (client.agent_engines.update/create), because
only that path writes the top-level spec.agent_card; the
vertexai.agent_engines.create/update proto (ReasoningEngineSpec) in this SDK
version has no agent_card field.

Related version conflict (mostly covered by #6274). We could not use
google.adk.a2a (A2aAgentExecutor): google-adk pins a2a-sdk>=0.3.4,<0.4,
but the Agent Engine A2A runtime and A2aAgent template are a2a-sdk 1.x /
protocol 1.0 (create_agent_card emits supported_interfaces at HTTP+JSON/1.0;
the handler is validate_version(PROTOCOL_VERSION_1_0)). So we installed
a2a-sdk[http-server]>=1.0,<2, avoided google.adk.a2a, and wrapped an ADK
core Runner in a hand-written a2a-sdk 1.x AgentExecutor. If #6274 lands,
that executor could be ADK's own — but the hybrid registration/deploy is still
needed.

Papercuts encountered (may be separate/docs):

1. A2A-Version: 1.0 header is mandatory. Missing → read as 0.3 and
   rejected (400 VERSION_NOT_SUPPORTED, "A2A version '0.3' is not supported by
   this handler. Expected version '1.0'."). The a2a-sdk ClientFactory sets it
   via setdefault, so SDK clients are fine; hand-rolled HTTP must add it.
2. Executor event ordering. Emitting a TaskStatusUpdateEvent
   (TaskUpdater.start_work()) before enqueuing the initial Task →
   500 INVALID_AGENT_RESPONSE ("Agent should enqueue Task before
   TaskStatusUpdateEvent event"). A clearer error / guardrail would help.
3. Default in-memory task store. A2aAgent defaults to InMemoryTaskStore,
   so a tasks/{id} GET after the request returns 404 TASK_NOT_FOUND
   (surprising on a multi-instance serverless runtime; message:send returns the
   completed task synchronously, so it's survivable).
4. adk api_server --a2a silently mounts no A2A routes (UnboundLocalError
   on json in get_fast_api_app). Filed separately — see adk api_server --a2a silently mounts no A2A routes: UnboundLocalError on json in get_fast_api_app #6332.

Request. Provide one of: an AdkApp(..., a2a=True, agent_card=...) option that
merges a2a_extension into register_operations() and wires an
A2aAgentExecutor around the app's Runner; or a supported composition helper
(e.g. to_a2a(app, keep_adk_ops=True)) / Agent-Engine template yielding a
deployable object with both surfaces; plus clearer runtime errors for papercuts
(1)–(2). The templates live in google-cloud-aiplatform while to_a2a /
A2aAgentExecutor live in adk-python, so the fix may span both repos; filing
here to start the conversation.

Minimal Reproduction Code:

from google.adk.agents import LlmAgent
from vertexai.agent_engines import AdkApp
from vertexai.agent_engines.templates.a2a import A2aAgent, create_agent_card
from a2a.types import AgentSkill

agent = LlmAgent(name="hello_agent", model="gemini-2.0-flash", instruction="hi")

adk = AdkApp(agent=agent)
print("AdkApp :", list(adk.register_operations().keys()))
# -> ['', 'async', 'stream', 'async_stream']

card = create_agent_card(
    agent_name="hello_agent", description="hi",
    skills=[AgentSkill(id="s", name="s", description="s", tags=[])],
)
a2a = A2aAgent(agent_card=card, agent_executor_builder=lambda: None)
print("A2aAgent:", list(a2a.register_operations().keys()))
# -> ['a2a_extension']

Also relevant: google-cloud-aiplatform 1.159.0, a2a-sdk 1.1.0 (protocol 1.0),
target Vertex AI Agent Engine (us-central1).

How often has this issue occurred?:

• Always (100%) — the templates always register only their own surface.