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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

eavanvalkenburg wants to merge 86 commits into microsoft:main from eavanvalkenburg:python_single_response

+11,305 −9,025

.github/workflows/python-merge-tests.yml

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -96,8 +96,7 @@ jobs:
  
            uses: ./.github/actions/azure-functions-integration-setup

            id: azure-functions-setup

          - name: Test with pytest

            timeout-minutes: 10

            run: uv run poe all-tests -n logical --dist loadfile --dist worksteal --timeout 900 --retries 3 --retry-delay 10

            run: uv run poe all-tests -n logical --dist loadfile --dist worksteal --timeout=120 --session-timeout=900 --timeout_method thread --retries 2 --retry-delay 5

            working-directory: ./python

          - name: Test core samples

            timeout-minutes: 10

    @@ -153,8 +152,8 @@ jobs:
  
              tenant-id: ${{ secrets.AZURE_TENANT_ID }}

              subscription-id: ${{ secrets.AZURE_SUBSCRIPTION_ID }}

          - name: Test with pytest

            timeout-minutes: 10

            run: uv run --directory packages/azure-ai poe integration-tests -n logical --dist loadfile --dist worksteal --timeout 300 --retries 3 --retry-delay 10

            timeout-minutes: 15

            run: uv run --directory packages/azure-ai poe integration-tests -n logical --dist loadfile --dist worksteal --timeout=120 --session-timeout=900 --timeout_method thread --retries 2 --retry-delay 5

            working-directory: ./python

          - name: Test Azure AI samples

            timeout-minutes: 10

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

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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

-Original file line number
+Diff line change
@@ Expand Up / @@ -38,6 +38,8 @@ @@
             "endregion",
             "entra",
             "faiss",
+            "finalizer",
+            "finalizers",
             "genai",
             "generativeai",
             "hnsw",
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -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.
@@ Expand Down @@

python/packages/a2a/agent_framework_a2a/_agent.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand All / @@ -32,10 +32,12 @@ @@
         BaseAgent,
         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"]
@@ Expand All / @@ -56,8 +58,7 @@ def _get_uri_data(uri: str) -> str: @@
         return match.group("base64_data")
-    @use_agent_instrumentation
-    class A2AAgent(BaseAgent):
+    class A2AAgent(AgentTelemetryLayer, BaseAgent):
         """Agent2Agent (A2A) protocol implementation.
         Wraps an A2A Client to connect the Agent Framework with external A2A-compliant agents
@@ Expand Down Expand Up / @@ -184,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 | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+            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)]
-            return AgentResponse.from_updates(updates)
+            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 | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+            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.
@@ Expand All / @@ -231,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)
@@ Expand All / @@ -244,7 +293,7 @@ async def run_stream( @@
                     contents = self._parse_contents_from_a2a(item.parts)
                     yield AgentResponseUpdate(
                         contents=contents,
-                        role="assistant" if item.role == A2ARole.agent else "user",
+                        role=Role.ASSISTANT if item.role == A2ARole.agent else Role.USER,
                         response_id=str(getattr(item, "message_id", uuid.uuid4())),
                         raw_representation=item,
                     )
@@ Expand All / @@ -268,7 +317,7 @@ async def run_stream( @@
                             # Empty task
                             yield AgentResponseUpdate(
                                 contents=[],
-                                role="assistant",
+                                role=Role.ASSISTANT,
                                 response_id=task.id,
                                 raw_representation=task,
                             )
@@ Expand Down Expand Up @@
                 contents = self._parse_contents_from_a2a(history_item.parts)
                 messages.append(
                     ChatMessage(
-                        role="assistant" if history_item.role == A2ARole.agent else "user",
+                        role=Role.ASSISTANT if history_item.role == A2ARole.agent else Role.USER,
                         contents=contents,
                         raw_representation=history_item,
                     )
@@ Expand All @@
             """Parse A2A Artifact into ChatMessage using part contents."""
             contents = self._parse_contents_from_a2a(artifact.parts)
             return ChatMessage(
-                role="assistant",
+                role=Role.ASSISTANT,
                 contents=contents,
                 raw_representation=artifact,
             )

python/packages/a2a/tests/test_a2a_agent.py

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -128,7 +128,7 @@ async def test_run_with_message_response(a2a_agent: A2AAgent, mock_a2a_client: M
  
        assert isinstance(response, AgentResponse)

        assert len(response.messages) == 1

        assert response.messages[0].role == "assistant"

        assert response.messages[0].role.value == "assistant"

        assert response.messages[0].text == "Hello from agent!"

        assert response.response_id == "msg-123"

        assert mock_a2a_client.call_count == 1

    @@ -143,7 +143,7 @@ async def test_run_with_task_response_single_artifact(a2a_agent: A2AAgent, mock_
  
        assert isinstance(response, AgentResponse)

        assert len(response.messages) == 1

        assert response.messages[0].role == "assistant"

        assert response.messages[0].role.value == "assistant"

        assert response.messages[0].text == "Generated report content"

        assert response.response_id == "task-456"

        assert mock_a2a_client.call_count == 1

    @@ -169,7 +169,7 @@ async def test_run_with_task_response_multiple_artifacts(a2a_agent: A2AAgent, mo
  
        # All should be assistant messages

        for message in response.messages:

            assert message.role == "assistant"

            assert message.role.value == "assistant"

        assert response.response_id == "task-789"

    @@ -232,7 +232,7 @@ def test_parse_messages_from_task_with_artifacts(a2a_agent: A2AAgent) -> None:
  
        assert len(result) == 2

        assert result[0].text == "Content 1"

        assert result[1].text == "Content 2"

        assert all(msg.role == "assistant" for msg in result)

        assert all(msg.role.value == "assistant" for msg in result)

    def test_parse_message_from_artifact(a2a_agent: A2AAgent) -> None:

    @@ -251,7 +251,7 @@ def test_parse_message_from_artifact(a2a_agent: A2AAgent) -> None:
  
        result = a2a_agent._parse_message_from_artifact(artifact)

        assert isinstance(result, ChatMessage)

        assert result.role == "assistant"

        assert result.role.value == "assistant"

        assert result.text == "Artifact content"

        assert result.raw_representation == artifact

    @@ -295,7 +295,7 @@ def test_prepare_message_for_a2a_with_error_content(a2a_agent: A2AAgent) -> None
  
        # Create ChatMessage with ErrorContent

        error_content = Content.from_error(message="Test error message")

        message = ChatMessage("user", [error_content])

        message = ChatMessage(role="user", contents=[error_content])

        # Convert to A2A message

        a2a_message = a2a_agent._prepare_message_for_a2a(message)

    @@ -310,7 +310,7 @@ def test_prepare_message_for_a2a_with_uri_content(a2a_agent: A2AAgent) -> None:
  
        # Create ChatMessage with UriContent

        uri_content = Content.from_uri(uri="http://example.com/file.pdf", media_type="application/pdf")

        message = ChatMessage("user", [uri_content])

        message = ChatMessage(role="user", contents=[uri_content])

        # Convert to A2A message

        a2a_message = a2a_agent._prepare_message_for_a2a(message)

    @@ -326,7 +326,7 @@ def test_prepare_message_for_a2a_with_data_content(a2a_agent: A2AAgent) -> None:
  
        # Create ChatMessage with DataContent (base64 data URI)

        data_content = Content.from_uri(uri="data:text/plain;base64,SGVsbG8gV29ybGQ=", media_type="text/plain")

        message = ChatMessage("user", [data_content])

        message = ChatMessage(role="user", contents=[data_content])

        # Convert to A2A message

        a2a_message = a2a_agent._prepare_message_for_a2a(message)

    @@ -340,26 +340,26 @@ def test_prepare_message_for_a2a_with_data_content(a2a_agent: A2AAgent) -> None:
  
    def test_prepare_message_for_a2a_empty_contents_raises_error(a2a_agent: A2AAgent) -> None:

        """Test _prepare_message_for_a2a with empty contents raises ValueError."""

        # Create ChatMessage with no contents

        message = ChatMessage("user", [])

        message = ChatMessage(role="user", contents=[])

        # Should raise ValueError for empty contents

        with raises(ValueError, match="ChatMessage.contents is empty"):

            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

        assert len(updates) == 1

        assert isinstance(updates[0], AgentResponseUpdate)

        assert updates[0].role == "assistant"

        assert updates[0].role.value == "assistant"

        assert len(updates[0].contents) == 1

        content = updates[0].contents[0]

python/packages/ag-ui/README.md

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -46,7 +46,7 @@ from agent_framework.ag_ui import AGUIChatClient
  
    async def main():

        async with AGUIChatClient(endpoint="http://localhost:8000/") as client:

            # Stream responses

            async for update in client.get_streaming_response("Hello!"):

            async for update in client.get_response("Hello!", stream=True):

                for content in update.contents:

                    if isinstance(content, TextContent):

                        print(content.text, end="", flush=True)

python/packages/ag-ui/ag_ui_tests/__init__.py

Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		# Copyright (c) Microsoft. All rights reserved.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

Diff view

Diff view

Uh oh!

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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

Are you sure you want to change the base?

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

Uh oh!

Diff view

Diff view

Uh oh!

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!