Python: Reorganize A2A samples and use package A2AExecutor

python/packages/a2a/agent_framework_a2a/_a2a_executor.py

@@ -2,6 +2,7 @@
 
 import base64
 import logging
+import uuid
 from asyncio import CancelledError
 from collections.abc import Mapping
 from functools import partial
@@ -181,9 +182,18 @@ async def _run_stream(self, query: Any, session: AgentSession, updater: TaskUpda
         """Run the agent in streaming mode and publish updates to the task updater."""
         response_stream = self._agent.run(query, session=session, stream=True, **self._run_kwargs)
         streamed_artifact_ids: set[str] = set()
+        # Generate a stable artifact ID for the entire stream so all chunks share the same ID.
+        # This ensures clients can coalesce streaming tokens into a single artifact/message
+        # per the A2A spec (TaskArtifactUpdateEvent with append=True on same artifactId).
+        default_artifact_id = str(uuid.uuid4())
         await (
             response_stream.with_transform_hook(
-                partial(self.handle_events, updater=updater, streamed_artifact_ids=streamed_artifact_ids)
+                partial(
+                    self.handle_events,
+                    updater=updater,
+                    streamed_artifact_ids=streamed_artifact_ids,
+                    default_artifact_id=default_artifact_id,
+                )
             )
         ).get_final_response()
 
@@ -199,7 +209,11 @@ async def _run(self, query: Any, session: AgentSession, updater: TaskUpdater) ->
             await self.handle_events(message, updater)
 
     async def handle_events(
-        self, item: Message | AgentResponseUpdate, updater: TaskUpdater, streamed_artifact_ids: set[str] | None = None
+        self,
+        item: Message | AgentResponseUpdate,
+        updater: TaskUpdater,
+        streamed_artifact_ids: set[str] | None = None,
+        default_artifact_id: str | None = None,
     ) -> None:
         """Convert agent response items (Messages or Updates) to A2A protocol events.
 
@@ -213,7 +227,10 @@ async def handle_events(
             item: The agent response item (Message or AgentResponseUpdate) to process.
             updater: The task updater to publish events to.
             streamed_artifact_ids: A set of artifact IDs that have already been streamed.
-                Used to prevent duplicate updates for the same artifact.
+                Used to track which artifacts need append=True on subsequent chunks.
+            default_artifact_id: A stable artifact ID to use when the item does not provide one.
+                This ensures all streaming chunks for a single response share the same artifact ID,
+                allowing clients to coalesce them into a single message.
 
         Example:
             .. code-block:: python
@@ -224,6 +241,7 @@ async def handle_events(
                         item: Message | AgentResponseUpdate,
                         updater: TaskUpdater,
                         streamed_artifact_ids: set[str] | None = None,
+                        default_artifact_id: str | None = None,
                     ) -> None:
                         # Custom logic to transform item contents
                         if item.role == "assistant" and item.contents:
@@ -260,19 +278,22 @@ async def handle_events(
 
         if parts:
             if isinstance(item, AgentResponseUpdate):
+                # Resolve artifact ID: use item's message_id if available, otherwise fall back
+                # to the stable default_artifact_id so all streaming chunks share the same ID.
+                artifact_id = item.message_id or default_artifact_id
                 # For streaming updates, we send TaskArtifactUpdateEvent via add_artifact
                 await updater.add_artifact(
                     parts=parts,
-                    artifact_id=item.message_id,
+                    artifact_id=artifact_id,
                     metadata=metadata,
                     append=(
                         True
-                        if streamed_artifact_ids is not None and item.message_id in (streamed_artifact_ids or set())
+                        if streamed_artifact_ids is not None and artifact_id in streamed_artifact_ids
                         else None
                     ),
                 )
-                if item.message_id and streamed_artifact_ids is not None:
-                    streamed_artifact_ids.add(item.message_id)
+                if artifact_id and streamed_artifact_ids is not None:
+                    streamed_artifact_ids.add(artifact_id)
             else:
                 # For final messages, we send TaskStatusUpdateEvent with 'working' state
                 await updater.update_status(

python/samples/02-agents/a2a/README.md

@@ -0,0 +1,54 @@
+# A2A Client Samples
+
+These samples demonstrate how to **consume** remote A2A-compliant agents using the Agent Framework's `A2AAgent` class.
+
+For hosting your own agents as A2A servers, see [`samples/04-hosting/a2a/`](../../04-hosting/a2a/).
+
+## Samples
+
+| Sample | Concept |
+|--------|---------|
+| [`agent_with_a2a.py`](agent_with_a2a.py) | Basic consumption — non-streaming and streaming |
+| [`a2a_agent_as_function_tools.py`](a2a_agent_as_function_tools.py) | Expose A2A skills as function tools for a host agent |
+| [`a2a_polling.py`](a2a_polling.py) | Poll a long-running task with continuation tokens |
+| [`a2a_stream_reconnection.py`](a2a_stream_reconnection.py) | Resume an interrupted stream via continuation token |
+| [`a2a_protocol_selection.py`](a2a_protocol_selection.py) | Configure preferred protocol bindings (JSONRPC, GRPC, HTTP+JSON) |
+
+## Prerequisites
+
+- A running A2A-compliant agent server (see `samples/04-hosting/a2a/` to start one)
+- Set `A2A_AGENT_HOST` environment variable to the server URL
+- For `a2a_agent_as_function_tools.py`: also set `FOUNDRY_PROJECT_ENDPOINT` and `FOUNDRY_MODEL`
+
+## Running
+
+```bash
+cd python/samples/02-agents/a2a
+
+# Start an A2A server in another terminal first:
+#   cd python/samples/04-hosting/a2a && uv run python a2a_server.py
+
+export A2A_AGENT_HOST="http://localhost:5001/"
+uv run python agent_with_a2a.py
+```
+
+## Key APIs
+
+```python
+from agent_framework.a2a import A2AAgent
+
+# Connect to a remote agent
+async with A2AAgent(url="http://localhost:5001/", agent_card=card) as agent:
+    # Non-streaming
+    response = await agent.run("Hello")
+
+    # Streaming
+    stream = agent.run("Hello", stream=True)
+    async for update in stream:
+        print(update.text)
+
+    # Background + polling
+    response = await agent.run("Long task", background=True)
+    while response.continuation_token:
+        response = await agent.poll_task(response.continuation_token)
+```

python/samples/04-hosting/a2a/a2a_agent_as_function_tools.py → python/samples/02-agents/a2a/a2a_agent_as_function_tools.py

@@ -33,7 +33,7 @@
 - Set FOUNDRY_MODEL to the model deployment name (e.g. gpt-4o)
 
 To run this sample:
-    cd python/samples/04-hosting/a2a
+    cd python/samples/02-agents/a2a
     uv run python a2a_agent_as_function_tools.py
 """
 

python/samples/02-agents/a2a/a2a_polling.py

@@ -0,0 +1,96 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+import httpx
+from a2a.client import A2ACardResolver
+from agent_framework.a2a import A2AAgent
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+A2A Polling for Task Completion
+
+This sample demonstrates how to poll a long-running A2A task for completion
+using continuation tokens. When `background=True`, the agent returns immediately
+with a continuation token that you can use to check progress later.
+
+Key concepts demonstrated:
+- Starting a background A2A task with `background=True`
+- Receiving a continuation token for in-progress tasks
+- Polling with `poll_task()` until the task reaches a terminal state
+
+This is the A2A equivalent of the .NET A2AAgent_PollingForTaskCompletion sample.
+
+Prerequisites:
+- Set A2A_AGENT_HOST to the URL of a running A2A server
+
+To run this sample:
+    cd python/samples/02-agents/a2a
+    uv run python a2a_polling.py
+"""
+
+
+async def main() -> None:
+    """Demonstrates polling a long-running A2A task for completion."""
+    a2a_agent_host = os.getenv("A2A_AGENT_HOST")
+    if not a2a_agent_host:
+        raise ValueError("A2A_AGENT_HOST environment variable is not set")
+
+    # 1. Resolve agent card and create agent.
+    async with httpx.AsyncClient(timeout=60.0) as http_client:
+        resolver = A2ACardResolver(httpx_client=http_client, base_url=a2a_agent_host)
+        agent_card = await resolver.get_agent_card()
+
+    async with A2AAgent(
+        name=agent_card.name,
+        agent_card=agent_card,
+        url=a2a_agent_host,
+    ) as agent:
+        # 2. Start a background task — the agent returns immediately.
+        print("Starting background task...")
+        response = await agent.run(
+            "Write a detailed research report on quantum computing advances in 2025",
+            background=True,
+        )
+
+        # 3. Check if we got a continuation token (task still in progress).
+        if response.continuation_token is None:
+            # Task completed immediately — no polling needed.
+            print("Task completed immediately:")
+            print(f"  {response.text}")
+            return
+
+        # 4. Poll until the task completes.
+        token = response.continuation_token
+        poll_count = 0
+        while token is not None:
+            poll_count += 1
+            print(f"  Poll #{poll_count} — task still in progress, waiting 2s...")
+            await asyncio.sleep(2)
+
+            response = await agent.poll_task(token)  # type: ignore[arg-type]
+            token = response.continuation_token
+
+        # 5. Task is done — print the final response.
+        print(f"\nTask completed after {poll_count} poll(s):")
+        print(f"  {response.text[:200]}...")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+
+"""
+Sample output:
+
+Starting background task...
+  Poll #1 — task still in progress, waiting 2s...
+  Poll #2 — task still in progress, waiting 2s...
+  Poll #3 — task still in progress, waiting 2s...
+
+Task completed after 3 poll(s):
+  Quantum computing has seen remarkable progress in 2025, with breakthroughs in...
+"""

python/samples/02-agents/a2a/a2a_protocol_selection.py

@@ -0,0 +1,84 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import os
+
+import httpx
+from a2a.client import A2ACardResolver
+from agent_framework.a2a import A2AAgent
+from dotenv import load_dotenv
+
+load_dotenv()
+
+"""
+A2A Protocol Selection
+
+This sample demonstrates how to configure which protocol binding the A2A client
+uses when connecting to a remote agent. The A2A specification defines three
+standard bindings: JSONRPC, GRPC, and HTTP+JSON. Agents declare their supported
+bindings in their AgentCard, and clients can express a preference.
+
+Key concepts demonstrated:
+- Configuring `supported_protocol_bindings` on A2AAgent
+- The client selects a binding that matches the remote agent's capabilities
+- Fallback behavior when preferred binding is unavailable
+
+This is the A2A equivalent of the .NET A2AAgent_ProtocolSelection sample.
+
+Prerequisites:
+- Set A2A_AGENT_HOST to the URL of a running A2A server
+
+To run this sample:
+    cd python/samples/02-agents/a2a
+    uv run python a2a_protocol_selection.py
+"""
+
+
+async def main() -> None:
+    """Demonstrates configuring A2A protocol binding preferences."""
+    a2a_agent_host = os.getenv("A2A_AGENT_HOST")
+    if not a2a_agent_host:
+        raise ValueError("A2A_AGENT_HOST environment variable is not set")
+
+    # 1. Resolve agent card to see what bindings are available.
+    async with httpx.AsyncClient(timeout=60.0) as http_client:
+        resolver = A2ACardResolver(httpx_client=http_client, base_url=a2a_agent_host)
+        agent_card = await resolver.get_agent_card()
+
+    print(f"Agent: {agent_card.name}")
+    print("Supported interfaces:")
+    for interface in agent_card.supported_interfaces:
+        print(f"  - {interface.protocol_binding} @ {interface.url}")
+
+    # 2. Create agent with explicit protocol binding preference.
+    #    The list is ordered by preference — the SDK will select the first
+    #    binding that matches a supported interface on the agent card.
+    #
+    #    This matters when a server exposes multiple interfaces (e.g. JSONRPC
+    #    on / and HTTP+JSON on /api/). If only one binding is available, the
+    #    client uses it regardless of your preference list.
+    async with A2AAgent(
+        name=agent_card.name,
+        agent_card=agent_card,
+        url=a2a_agent_host,
+        supported_protocol_bindings=["HTTP+JSON", "JSONRPC"],
+    ) as agent:
+        print("\nConfigured bindings: ['HTTP+JSON', 'JSONRPC']")
+        response = await agent.run("Tell me a short joke")
+        print(f"Response: {response.text}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
+
+"""
+Sample output:
+
+Agent: PolicyAgent
+Supported interfaces:
+  - JSONRPC @ http://localhost:5001/
+
+Configured bindings: ['HTTP+JSON', 'JSONRPC']
+Response: Here's a short joke for you...
+"""