Python: [BREAKING] Moved to a single get_response and run API

docs/decisions/0012-python-typeddict-options.md

@@ -126,4 +126,4 @@ response = await client.get_response(
 
 Chosen option: **"Option 2: TypedDict with Generic Type Parameters"**, because it provides full type safety, excellent IDE support with autocompletion, and allows users to extend provider-specific options for their use cases. Extended this Generic to ChatAgents in order to also properly type the options used in agent construction and run methods.
 
-See [typed_options.py](../../python/samples/getting_started/chat_client/typed_options.py) for a complete example demonstrating the usage of typed options with custom extensions.
+See [typed_options.py](../../python/samples/concepts/typed_options.py) for a complete example demonstrating the usage of typed options with custom extensions.

python/.cspell.json

@@ -38,6 +38,8 @@
         "endregion",
         "entra",
         "faiss",
+        "finalizer",
+        "finalizers",
         "genai",
         "generativeai",
         "hnsw",

python/.github/instructions/python.instructions.md

@@ -12,7 +12,7 @@ applyTo: '**/agent-framework/python/**'
 - Do not use `Optional`; use `Type | None` instead.
 - Before running any commands to execute or test the code, ensure that all problems, compilation errors, and warnings are resolved.
 - When formatting files, format only the files you changed or are currently working on; do not format the entire codebase.
-- Do not mark new tests with `@pytest.mark.asyncio`.
+- Do not mark new tests with `@pytest.mark.asyncio`, they are marked automatically, so you can just set the test to `async def`.
 - If you need debug information to understand an issue, use print statements as needed and remove them when testing is complete.
 - Avoid adding excessive comments.
 - When working with samples, make sure to update the associated README files with the latest information. These files are usually located in the same folder as the sample or in one of its parent folders.

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -4,8 +4,8 @@
 import json
 import re
 import uuid
-from collections.abc import AsyncIterable, Sequence
-from typing import Any, Final, cast
+from collections.abc import AsyncIterable, Awaitable, Sequence
+from typing import Any, Final, Literal, cast, overload
 
 import httpx
 from a2a.client import Client, ClientConfig, ClientFactory, minimal_agent_card
@@ -29,14 +29,15 @@
     AgentResponse,
     AgentResponseUpdate,
     AgentThread,
-    BaseAgent,
+    BareAgent,
     ChatMessage,
     Content,
+    ResponseStream,
     Role,
     normalize_messages,
     prepend_agent_framework_to_user_agent,
 )
-from agent_framework.observability import use_agent_instrumentation
+from agent_framework.observability import AgentTelemetryLayer
 
 __all__ = ["A2AAgent"]
 
@@ -57,13 +58,12 @@ def _get_uri_data(uri: str) -> str:
     return match.group("base64_data")
 
 
-@use_agent_instrumentation
-class A2AAgent(BaseAgent):
+class A2AAgent(AgentTelemetryLayer, BareAgent):
     """Agent2Agent (A2A) protocol implementation.
 
     Wraps an A2A Client to connect the Agent Framework with external A2A-compliant agents
     via HTTP/JSON-RPC. Converts framework ChatMessages to A2A Messages on send, and converts
-    A2A responses (Messages/Tasks) back to framework types. Inherits BaseAgent capabilities
+    A2A responses (Messages/Tasks) back to framework types. Inherits BareAgent capabilities
     while managing the underlying A2A protocol communication.
 
     Can be initialized with a URL, AgentCard, or existing A2A Client instance.
@@ -99,7 +99,7 @@ def __init__(
             timeout: Request timeout configuration. Can be a float (applied to all timeout components),
                 httpx.Timeout object (for full control), or None (uses 10.0s connect, 60.0s read,
                 10.0s write, 5.0s pool - optimized for A2A operations).
-            kwargs: any additional properties, passed to BaseAgent.
+            kwargs: any additional properties, passed to BareAgent.
         """
         super().__init__(id=id, name=name, description=description, **kwargs)
         self._http_client: httpx.AsyncClient | None = http_client
@@ -185,44 +185,92 @@ async def __aexit__(
         if self._http_client is not None and self._close_http_client:
             await self._http_client.aclose()
 
-    async def run(
+    @overload
+    def run(
         self,
         messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
+        stream: Literal[False] = ...,
         thread: AgentThread | None = None,
         **kwargs: Any,
-    ) -> AgentResponse:
+    ) -> Awaitable[AgentResponse[Any]]: ...
+
+    @overload
+    def run(
+        self,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
+        *,
+        stream: Literal[True],
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
+
+    def run(
+        self,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
+        *,
+        stream: bool = False,
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
         """Get a response from the agent.
 
         This method returns the final result of the agent's execution
-        as a single AgentResponse object. The caller is blocked until
-        the final result is available.
+        as a single AgentResponse object when stream=False. When stream=True,
+        it returns a ResponseStream that yields AgentResponseUpdate objects.
 
         Args:
             messages: The message(s) to send to the agent.
 
         Keyword Args:
+            stream: Whether to stream the response. Defaults to False.
             thread: The conversation thread associated with the message(s).
             kwargs: Additional keyword arguments.
 
         Returns:
-            An agent response item.
+            When stream=False: An Awaitable[AgentResponse].
+            When stream=True: A ResponseStream of AgentResponseUpdate items.
         """
+        if stream:
+            return self._run_stream_impl(messages=messages, thread=thread, **kwargs)
+        return self._run_impl(messages=messages, thread=thread, **kwargs)
+
+    async def _run_impl(
+        self,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
+        *,
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> AgentResponse[Any]:
+        """Non-streaming implementation of run."""
         # Collect all updates and use framework to consolidate updates into response
-        updates = [update async for update in self.run_stream(messages, thread=thread, **kwargs)]
+        updates: list[AgentResponseUpdate] = []
+        async for update in self._stream_updates(messages, thread=thread, **kwargs):
+            updates.append(update)
         return AgentResponse.from_agent_run_response_updates(updates)
 
-    async def run_stream(
+    def _run_stream_impl(
         self,
         messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
-    ) -> AsyncIterable[AgentResponseUpdate]:
-        """Run the agent as a stream.
+    ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
+        """Streaming implementation of run."""
+
+        def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]:
+            return AgentResponse.from_agent_run_response_updates(list(updates))
+
+        return ResponseStream(self._stream_updates(messages, thread=thread, **kwargs), finalizer=_finalize)
 
-        This method will return the intermediate steps and final results of the
-        agent's execution as a stream of AgentResponseUpdate objects to the caller.
+    async def _stream_updates(
+        self,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
+        *,
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> AsyncIterable[AgentResponseUpdate]:
+        """Internal method to stream updates from the A2A agent.
 
         Args:
             messages: The message(s) to send to the agent.
@@ -232,10 +280,10 @@ async def run_stream(
             kwargs: Additional keyword arguments.
 
         Yields:
-            An agent response item.
+            AgentResponseUpdate items from the A2A agent.
         """
-        messages = normalize_messages(messages)
-        a2a_message = self._prepare_message_for_a2a(messages[-1])
+        normalized_messages = normalize_messages(messages)
+        a2a_message = self._prepare_message_for_a2a(normalized_messages[-1])
 
         response_stream = self.client.send_message(a2a_message)
 

python/packages/a2a/tests/test_a2a_agent.py

@@ -348,13 +348,13 @@ def test_prepare_message_for_a2a_empty_contents_raises_error(a2a_agent: A2AAgent
         a2a_agent._prepare_message_for_a2a(message)
 
 
-async def test_run_stream_with_message_response(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
-    """Test run_stream() method with immediate Message response."""
+async def test_run_streaming_with_message_response(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
+    """Test run(stream=True) method with immediate Message response."""
     mock_a2a_client.add_message_response("msg-stream-123", "Streaming response from agent!", "agent")
 
     # Collect streaming updates
     updates: list[AgentResponseUpdate] = []
-    async for update in a2a_agent.run_stream("Hello agent"):
+    async for update in a2a_agent.run("Hello agent", stream=True):
         updates.append(update)
 
     # Verify streaming response