Python: Address A2A channel CI failures and review feedback

- Fix HostAgentExecutor streaming artifact coalescing by using one stable
  per-task artifact id, even when update.message_id is missing or varies
- Add streaming fallback to text updates when contents are empty
- Fix workflow result projection in non-streaming mode by consuming
  get_outputs() values in addition to response.messages
- Add _value_to_parts helper to convert workflow outputs and fallback
  values into A2A Parts
- Expand tests to validate stable artifact ids and workflow output
  projection from HostedRunResult[WorkflowRunResult]
- Harden A2A test fakes for strict typing (agent protocol shape,
  context/request casts, event queue override)
- Add package-local test dependency group with uvicorn to make test
  execution self-contained

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: eavanvalkenburg

python/packages/hosting-a2a/agent_framework_hosting_a2a/_executor.py

@@ -22,6 +22,7 @@
 from a2a.server.tasks import TaskUpdater
 from a2a.types import Part, Task, TaskState
 from agent_framework import Content
+from agent_framework import Message as AFMessage
 from agent_framework_hosting import (
     ChannelContext,
     ChannelIdentity,
@@ -59,6 +60,17 @@ def _contents_to_parts(contents: list[Content]) -> list[Part]:
     return parts
 
 
+def _value_to_parts(value: Any) -> list[Part]:
+    """Convert workflow outputs and fallback values into A2A parts."""
+    if isinstance(value, Content):
+        return _contents_to_parts([value])
+    if isinstance(value, AFMessage):
+        return _contents_to_parts(list(value.contents))
+    if isinstance(value, str):
+        return [Part(text=value)]
+    return [Part(text=str(value))]
+
+
 class HostAgentExecutor(AgentExecutor):
     """A2A executor that drives the hosted target through :class:`ChannelContext`."""
 
@@ -158,11 +170,13 @@ async def _run(self, request: ChannelRequest, updater: TaskUpdater, *, protocol_
         )
         response: Any = result.result
         messages: list[Any] = list(getattr(response, "messages", None) or [])
+        get_outputs = cast("Any", getattr(response, "get_outputs", None))
+        if callable(get_outputs):
+            messages.extend(cast("list[Any]", get_outputs()))
         for message in messages:
             if getattr(message, "role", None) == "user":
                 continue
-            contents: list[Content] = list(getattr(message, "contents", None) or [])
-            parts = _contents_to_parts(contents)
+            parts = _value_to_parts(message)
             if parts:
                 await updater.update_status(
                     state=TaskState.TASK_STATE_WORKING,
@@ -171,7 +185,8 @@ async def _run(self, request: ChannelRequest, updater: TaskUpdater, *, protocol_
 
     async def _run_stream(self, request: ChannelRequest, updater: TaskUpdater, *, protocol_request: Any) -> None:
         """Streaming: publish incremental updates as task artifacts."""
-        streamed_ids: set[str] = set()
+        stream_artifact_id = f"{request.attributes.get('task_id', 'stream')}:stream"
+        appended = False
         stream = await self._ctx.run_stream(
             request,
             run_hook=self._run_hook,
@@ -182,14 +197,15 @@ async def _run_stream(self, request: ChannelRequest, updater: TaskUpdater, *, pr
         async for update in stream:
             contents: list[Content] = list(getattr(update, "contents", None) or [])
             parts = _contents_to_parts(contents)
+            text = getattr(update, "text", None)
+            if not parts and isinstance(text, str) and text:
+                parts = [Part(text=text)]
             if not parts:
                 continue
-            message_id: str | None = getattr(update, "message_id", None)
             await updater.add_artifact(
                 parts=parts,
-                artifact_id=message_id,
-                append=True if message_id is not None and message_id in streamed_ids else None,
+                artifact_id=stream_artifact_id,
+                append=True if appended else None,
             )
-            if message_id is not None:
-                streamed_ids.add(message_id)
+            appended = True
         await stream.get_final_response()

python/packages/hosting-a2a/pyproject.toml

@@ -79,3 +79,8 @@ cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_a2a --cov-repor
 [build-system]
 requires = ["flit-core >= 3.11,<4.0"]
 build-backend = "flit_core.buildapi"
+
+[dependency-groups]
+dev = [
+    "uvicorn[standard]>=0.34.0",
+]

python/packages/hosting-a2a/tests/hosting_a2a/test_channel.py

@@ -8,7 +8,7 @@
 from collections.abc import AsyncIterator, Awaitable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, cast
 
 import pytest
 import uvicorn
@@ -126,7 +126,6 @@ def __init__(self) -> None:
 
     async def enqueue_event(self, event: Any) -> None:
         self.events.append(event)
-        await super().enqueue_event(event)
 
 
 class _FakeRequestContext:
@@ -147,13 +146,20 @@ def get_user_input(self) -> str:
 
 
 class _HostedAgent:
-    name = "HostedAssistant"
-    description = "A hosted test assistant."
+    id = "hosted-agent"
+    name: str | None = "HostedAssistant"
+    description: str | None = "A hosted test assistant."
 
     async def run(self, messages: Any = None, *, stream: bool = False, **_kwargs: Any) -> AgentResponse[Any]:
         text = messages.text if isinstance(messages, AFMessage) else str(messages)
         return AgentResponse(messages=[AFMessage(role="assistant", contents=[Content.from_text(text=f"host: {text}")])])
 
+    def create_session(self, *, session_id: str | None = None) -> Any:
+        return {"session_id": session_id}
+
+    def get_session(self, service_session_id: str, *, session_id: str | None = None) -> Any:
+        return {"service_session_id": service_session_id, "session_id": session_id}
+
 
 @asynccontextmanager
 async def _serve_app(app: ASGIApp, *, port: int) -> AsyncIterator[str]:
@@ -182,6 +188,18 @@ def _status_states(events: list[Any]) -> list[int]:
     return states
 
 
+def _status_texts(events: list[Any]) -> list[str]:
+    texts: list[str] = []
+    for event in events:
+        status = getattr(event, "status", None)
+        message = getattr(status, "message", None)
+        for part in cast("list[Any]", getattr(message, "parts", None) or []):
+            text = getattr(part, "text", None)
+            if isinstance(text, str):
+                texts.append(text)
+    return texts
+
+
 # --------------------------------------------------------------------------- #
 # A2AChannel tests                                                             #
 # --------------------------------------------------------------------------- #
@@ -195,7 +213,7 @@ def test_default_name_and_root_path() -> None:
 
 def test_build_agent_card_defaults_from_target() -> None:
     channel = A2AChannel(url="https://example.com/")
-    card = channel._build_agent_card(_FakeContext())  # type: ignore[arg-type]
+    card = channel._build_agent_card(cast(Any, _FakeContext()))
     assert card.name == "Assistant"
     assert card.description == "A helpful assistant."
     assert card.capabilities.streaming is True
@@ -208,21 +226,21 @@ def test_build_agent_card_accepts_supported_interfaces() -> None:
         AgentInterface(url="https://example.com/grpc", protocol_binding="GRPC"),
     ]
     channel = A2AChannel(supported_interfaces=interfaces)
-    card = channel._build_agent_card(_FakeContext())  # type: ignore[arg-type]
+    card = channel._build_agent_card(cast(Any, _FakeContext()))
     assert card.supported_interfaces == interfaces
 
 
 def test_build_agent_card_override_wins() -> None:
     custom = AgentCard(name="Custom", description="custom card", version="9.9.9")
     channel = A2AChannel(agent_card=custom)
-    card = channel._build_agent_card(_FakeContext())  # type: ignore[arg-type]
+    card = channel._build_agent_card(cast(Any, _FakeContext()))
     assert card.name == "Custom"
     assert card.version == "9.9.9"
 
 
 def test_contribute_returns_card_and_jsonrpc_routes() -> None:
     channel = A2AChannel(url="https://example.com/")
-    contribution = channel.contribute(_FakeContext())  # type: ignore[arg-type]
+    contribution = channel.contribute(cast(Any, _FakeContext()))
     assert isinstance(contribution, ChannelContribution)
     paths = {getattr(r, "path", None) for r in contribution.routes}
     assert "/.well-known/agent-card.json" in paths
@@ -236,11 +254,11 @@ def test_contribute_returns_card_and_jsonrpc_routes() -> None:
 
 async def test_execute_routes_through_host_and_completes() -> None:
     ctx = _FakeContext(reply="hi back")
-    executor = HostAgentExecutor(ctx, channel_name="a2a", streaming=False)  # type: ignore[arg-type]
+    executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=False)
     queue = _RecordingEventQueue()
     request_context = _FakeRequestContext(context_id="conv-1", text="hello")
 
-    await executor.execute(request_context, queue)  # type: ignore[arg-type]
+    await executor.execute(cast(Any, request_context), queue)
 
     # Routed through the host with the context id mapped onto the session.
     assert len(ctx.requests) == 1
@@ -257,31 +275,34 @@ async def test_execute_routes_through_host_and_completes() -> None:
 
 async def test_execute_streaming_emits_artifacts() -> None:
     ctx = _FakeContext(chunks=["foo", "bar"])
-    executor = HostAgentExecutor(ctx, channel_name="a2a", streaming=True)  # type: ignore[arg-type]
+    executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=True)
     queue = _RecordingEventQueue()
     request_context = _FakeRequestContext(context_id="conv-2", text="hello")
 
-    await executor.execute(request_context, queue)  # type: ignore[arg-type]
+    await executor.execute(cast(Any, request_context), queue)
 
     artifact_events = [e for e in queue.events if getattr(e, "artifact", None)]
     assert artifact_events, "expected at least one artifact update event"
+    artifact_ids = {getattr(getattr(e, "artifact", None), "artifact_id", None) for e in artifact_events}
+    assert len(artifact_ids) == 1
+    assert None not in artifact_ids
     assert ctx.requests[0].stream is True
     assert TaskState.TASK_STATE_COMPLETED in _status_states(queue.events)
 
 
 async def test_execute_requires_context_id() -> None:
     ctx = _FakeContext()
-    executor = HostAgentExecutor(ctx, channel_name="a2a")  # type: ignore[arg-type]
+    executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a")
     queue = _RecordingEventQueue()
     request_context = _FakeRequestContext(context_id="x", text="hello")
-    request_context.context_id = None  # type: ignore[assignment]
+    cast(Any, request_context).context_id = None
 
     with pytest.raises(ValueError, match="Context ID"):
-        await executor.execute(request_context, queue)  # type: ignore[arg-type]
+        await executor.execute(cast(Any, request_context), queue)
 
 
 async def test_a2a_agent_can_call_hosted_channel(unused_tcp_port: int) -> None:
-    host = AgentFrameworkHost(target=_HostedAgent(), channels=[A2AChannel(streaming=False)])
+    host = AgentFrameworkHost(target=cast(Any, _HostedAgent()), channels=[A2AChannel(streaming=False)])
 
     async with (
         _serve_app(host.app, port=unused_tcp_port) as base_url,
@@ -295,6 +316,36 @@ async def test_a2a_agent_can_call_hosted_channel(unused_tcp_port: int) -> None:
     assert response.messages[0].text == "host: hello"
 
 
+async def test_execute_projects_workflow_outputs() -> None:
+    class _WorkflowResult:
+        value = None
+
+        def get_outputs(self) -> list[AFMessage]:
+            return [AFMessage(role="assistant", contents=[Content.from_text("workflow output")])]
+
+    class _WorkflowContext(_FakeContext):
+        async def run(
+            self,
+            request: ChannelRequest,
+            *,
+            run_hook: Any | None = None,
+            protocol_request: Any | None = None,
+            response_hook: Any | None = None,
+            channel_name: str | None = None,
+        ) -> HostedRunResult[Any]:
+            self.requests.append(request)
+            return HostedRunResult(_WorkflowResult())
+
+    ctx = _WorkflowContext()
+    executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=False)
+    queue = _RecordingEventQueue()
+    request_context = _FakeRequestContext(context_id="conv-workflow", text="hello")
+
+    await executor.execute(cast(Any, request_context), queue)
+
+    assert "workflow output" in _status_texts(queue.events)
+
+
 def test_contents_to_parts_conversion() -> None:
     from agent_framework_hosting_a2a._executor import _contents_to_parts