Merge branch 'main' into openai_response_agent_completeness

python/packages/main/agent_framework/openai/_responses_client.py

@@ -579,10 +579,10 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                                     TextContent(text=message_content.refusal, raw_representation=message_content)
                                 )
                 case "reasoning":  # ResponseOutputReasoning
-                    if item.content:
+                    if hasattr(item, "content") and item.content:
                         for index, reasoning_content in enumerate(item.content):
                             additional_properties = None
-                            if item.summary and index < len(item.summary):
+                            if hasattr(item, "summary") and item.summary and index < len(item.summary):
                                 additional_properties = {"summary": item.summary[index]}
                             contents.append(
                                 TextReasoningContent(
@@ -592,7 +592,7 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                                 )
                             )
                 case "code_interpreter_call":  # ResponseOutputCodeInterpreterCall
-                    if item.outputs:
+                    if hasattr(item, "outputs") and item.outputs:
                         for code_output in item.outputs:
                             if code_output.type == "logs":
                                 contents.append(TextContent(text=code_output.logs, raw_representation=item))
@@ -605,16 +605,16 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                                         media_type="image",
                                     )
                                 )
-                    elif item.code:
+                    elif hasattr(item, "code") and item.code:
                         # fallback if no output was returned is the code:
                         contents.append(TextContent(text=item.code, raw_representation=item))
                 case "function_call":  # ResponseOutputFunctionCall
                     contents.append(
                         FunctionCallContent(
-                            call_id=item.call_id if item.call_id else "",
-                            name=item.name,
-                            arguments=item.arguments,
-                            additional_properties={"fc_id": item.id},
+                            call_id=item.call_id if hasattr(item, "call_id") and item.call_id else "",
+                            name=item.name if hasattr(item, "name") else "",
+                            arguments=item.arguments if hasattr(item, "arguments") else "",
+                            additional_properties={"fc_id": item.id} if hasattr(item, "id") else {},
                             raw_representation=item,
                         )
                     )
@@ -739,6 +739,18 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
             case "response.output_text.delta":
                 contents.append(TextContent(text=event.delta, raw_representation=event))
                 metadata.update(self._get_metadata_from_response(event))
+            case "response.reasoning_text.delta":
+                contents.append(TextReasoningContent(text=event.delta, raw_representation=event))
+                metadata.update(self._get_metadata_from_response(event))
+            case "response.reasoning_text.done":
+                contents.append(TextReasoningContent(text=event.text, raw_representation=event))
+                metadata.update(self._get_metadata_from_response(event))
+            case "response.reasoning_summary_text.delta":
+                contents.append(TextReasoningContent(text=event.delta, raw_representation=event))
+                metadata.update(self._get_metadata_from_response(event))
+            case "response.reasoning_summary_text.done":
+                contents.append(TextReasoningContent(text=event.text, raw_representation=event))
+                metadata.update(self._get_metadata_from_response(event))
             case "response.completed":
                 conversation_id = event.response.id if chat_options.store is True else None
                 model = event.response.model
@@ -779,7 +791,7 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                             )
                         )
                     case "code_interpreter_call":  # ResponseOutputCodeInterpreterCall
-                        if event_item.outputs:
+                        if hasattr(event_item, "outputs") and event_item.outputs:
                             for code_output in event_item.outputs:
                                 if code_output.type == "logs":
                                     contents.append(TextContent(text=code_output.logs, raw_representation=event_item))
@@ -792,14 +804,18 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                                             media_type="image",
                                         )
                                     )
-                        elif event_item.code:
+                        elif hasattr(event_item, "code") and event_item.code:
                             # fallback if no output was returned is the code:
                             contents.append(TextContent(text=event_item.code, raw_representation=event_item))
                     case "reasoning":  # ResponseOutputReasoning
-                        if event_item.content:
+                        if hasattr(event_item, "content") and event_item.content:
                             for index, reasoning_content in enumerate(event_item.content):
                                 additional_properties = None
-                                if event_item.summary and index < len(event_item.summary):
+                                if (
+                                    hasattr(event_item, "summary")
+                                    and event_item.summary
+                                    and index < len(event_item.summary)
+                                ):
                                     additional_properties = {"summary": event_item.summary[index]}
                                 contents.append(
                                     TextReasoningContent(

python/packages/main/agent_framework/workflow/__init__.py

@@ -56,6 +56,7 @@ _IMPORTS = [
     "PlanReviewRequest",
     "RequestInfoEvent",
     "StandardMagenticManager",
+    "ConcurrentBuilder",
 ]
 
 

python/packages/main/agent_framework/workflow/__init__.pyi

@@ -8,6 +8,7 @@ from agent_framework_workflow import (
     AgentRunUpdateEvent,
     Case,
     CheckpointStorage,
+    ConcurrentBuilder,
     Default,
     Executor,
     ExecutorCompletedEvent,
@@ -56,6 +57,7 @@ __all__ = [
     "AgentRunUpdateEvent",
     "Case",
     "CheckpointStorage",
+    "ConcurrentBuilder",
     "Default",
     "Executor",
     "ExecutorCompletedEvent",

python/packages/main/pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "azure-monitor-opentelemetry>=1.7.0",
     "azure-monitor-opentelemetry-exporter>=1.0.0b41",
     "opentelemetry-exporter-otlp-proto-grpc>=1.36.0",
+    "opentelemetry-semantic-conventions-ai>=0.4.13"
 ]
 
 [project.optional-dependencies]

python/packages/main/tests/openai/test_openai_responses_client.py

@@ -1494,6 +1494,7 @@ def test_service_response_exception_includes_original_error_details() -> None:
     assert original_error_message in exception_message
 
 
+
 def test_get_streaming_response_with_response_format() -> None:
     """Test get_streaming_response with response_format."""
     client = OpenAIResponsesClient(ai_model_id="test-model", api_key="test-key")

python/packages/workflow/agent_framework_workflow/__init__.py

@@ -9,6 +9,7 @@ from ._checkpoint import (
     InMemoryCheckpointStorage,
     WorkflowCheckpoint,
 )
+from ._concurrent import ConcurrentBuilder
 from ._const import (
     DEFAULT_MAX_ITERATIONS,
 )
@@ -93,6 +94,7 @@ __all__ = [
     "AgentRunUpdateEvent",
     "Case",
     "CheckpointStorage",
+    "ConcurrentBuilder",
     "Default",
     "EdgeDuplicationError",
     "Executor",

python/packages/workflow/agent_framework_workflow/_concurrent.py

@@ -0,0 +1,304 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import inspect
+import logging
+from collections.abc import Callable, Sequence
+from typing import Any
+
+from agent_framework import AgentProtocol, ChatMessage, Role
+
+from ._events import WorkflowCompletedEvent
+from ._executor import AgentExecutorRequest, AgentExecutorResponse, Executor, handler
+from ._workflow import Workflow, WorkflowBuilder
+from ._workflow_context import WorkflowContext
+
+logger = logging.getLogger(__name__)
+
+"""Concurrent builder for agent-only fan-out/fan-in workflows.
+
+This module provides a high-level, agent-focused API to quickly assemble a
+parallel workflow with:
+- a default dispatcher that broadcasts the input to all agent participants
+- a default aggregator that combines all agent conversations and completes the workflow
+
+Notes:
+- Participants should be AgentProtocol instances or Executors.
+- A custom aggregator can be provided as:
+  - an Executor instance (it should handle list[AgentExecutorResponse] and add a WorkflowCompletedEvent), or
+  - a callback function with signature:
+        def cb(results: list[AgentExecutorResponse]) -> Any | None
+        def cb(results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None
+    If the callback returns a non-None value, it is sent as the data of a WorkflowCompletedEvent.
+    If it returns None, the callback may have already emitted a completion event via ctx.
+"""
+
+
+class _DispatchToAllParticipants(Executor):
+    """Broadcasts input to all downstream participants (via fan-out edges)."""
+
+    @handler
+    async def from_request(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
+        # No explicit target: edge routing delivers to all connected participants.
+        await ctx.send_message(request)
+
+    @handler
+    async def from_str(self, prompt: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
+        request = AgentExecutorRequest(messages=[ChatMessage(Role.USER, text=prompt)], should_respond=True)
+        await ctx.send_message(request)
+
+    @handler
+    async def from_message(self, message: ChatMessage, ctx: WorkflowContext[AgentExecutorRequest]) -> None:  # type: ignore[name-defined]
+        request = AgentExecutorRequest(messages=[message], should_respond=True)
+        await ctx.send_message(request)
+
+    @handler
+    async def from_messages(self, messages: list[ChatMessage], ctx: WorkflowContext[AgentExecutorRequest]) -> None:  # type: ignore[name-defined]
+        request = AgentExecutorRequest(messages=list(messages), should_respond=True)
+        await ctx.send_message(request)
+
+
+class _AggregateAgentConversations(Executor):
+    """Aggregates agent responses and completes with combined ChatMessages.
+
+    Emits a list[ChatMessage] shaped as:
+      [ single_user_prompt?, agent1_final_assistant, agent2_final_assistant, ... ]
+
+    - Extracts a single user prompt (first user message seen across results).
+    - For each result, selects the final assistant message (prefers agent_run_response.messages).
+    - Avoids duplicating the same user message per agent.
+    """
+
+    @handler
+    async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> None:
+        if not results:
+            logger.error("Concurrent aggregator received empty results list")
+            raise ValueError("Aggregation failed: no results provided")
+
+        def _is_role(msg: Any, role: Role) -> bool:
+            r = getattr(msg, "role", None)
+            if r is None:
+                return False
+            # Normalize both r and role to lowercase strings for comparison
+            r_str = str(r).lower() if isinstance(r, str) or hasattr(r, "__str__") else r
+            role_str = getattr(role, "value", None)
+            if role_str is None:
+                role_str = str(role)
+            role_str = role_str.lower()
+            return r_str == role_str
+
+        prompt_message: ChatMessage | None = None
+        assistant_replies: list[ChatMessage] = []
+
+        for r in results:
+            resp_messages = list(getattr(r.agent_run_response, "messages", []) or [])
+            conv = r.full_conversation if r.full_conversation is not None else resp_messages
+
+            logger.debug(
+                f"Aggregating executor {getattr(r, 'executor_id', '<unknown>')}: "
+                f"{len(resp_messages)} response msgs, {len(conv)} conversation msgs"
+            )
+
+            # Capture a single user prompt (first encountered across any conversation)
+            if prompt_message is None:
+                found_user = next((m for m in conv if _is_role(m, Role.USER)), None)
+                if found_user is not None:
+                    prompt_message = found_user
+
+            # Pick the final assistant message from the response; fallback to conversation search
+            final_assistant = next((m for m in reversed(resp_messages) if _is_role(m, Role.ASSISTANT)), None)
+            if final_assistant is None:
+                final_assistant = next((m for m in reversed(conv) if _is_role(m, Role.ASSISTANT)), None)
+
+            if final_assistant is not None:
+                assistant_replies.append(final_assistant)
+            else:
+                logger.warning(
+                    f"No assistant reply found for executor {getattr(r, 'executor_id', '<unknown>')}; skipping"
+                )
+
+        if not assistant_replies:
+            logger.error(f"Aggregation failed: no assistant replies found across {len(results)} results")
+            raise RuntimeError("Aggregation failed: no assistant replies found")
+
+        output: list[ChatMessage] = []
+        if prompt_message is not None:
+            output.append(prompt_message)
+        else:
+            logger.warning("No user prompt found in any conversation; emitting assistants only")
+        output.extend(assistant_replies)
+
+        await ctx.add_event(WorkflowCompletedEvent(data=output))
+
+
+class _CallbackAggregator(Executor):
+    """Wraps a Python callback as an aggregator.
+
+    Accepts either an async or sync callback with one of the signatures:
+      - (results: list[AgentExecutorResponse]) -> Any | None
+      - (results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None
+
+    Notes:
+    - Async callbacks are awaited directly.
+    - Sync callbacks are executed via asyncio.to_thread to avoid blocking the event loop.
+    - If the callback returns a non-None value, it is wrapped in a WorkflowCompletedEvent.
+    """
+
+    def __init__(self, callback: Callable[..., Any], id: str | None = None) -> None:
+        super().__init__(id)
+        self._callback = callback
+        self._param_count = len(inspect.signature(callback).parameters)
+
+    @handler
+    async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> None:
+        # Call according to provided signature, always non-blocking for sync callbacks
+        if self._param_count >= 2:
+            if inspect.iscoroutinefunction(self._callback):
+                ret = await self._callback(results, ctx)  # type: ignore[misc]
+            else:
+                ret = await asyncio.to_thread(self._callback, results, ctx)
+        else:
+            if inspect.iscoroutinefunction(self._callback):
+                ret = await self._callback(results)  # type: ignore[misc]
+            else:
+                ret = await asyncio.to_thread(self._callback, results)
+
+        # If the callback returned a value, finalize the workflow with it
+        if ret is not None:
+            await ctx.add_event(WorkflowCompletedEvent(ret))
+
+
+class ConcurrentBuilder:
+    r"""High-level builder for concurrent agent workflows.
+
+    - `participants([...])` accepts a list of AgentProtocol (recommended) or Executor.
+    - `build()` wires: dispatcher -> fan-out -> participants -> fan-in -> aggregator.
+    - `with_custom_aggregator(...)` overrides the default aggregator with an Executor or callback.
+
+    Usage:
+    ```python
+    from agent_framework.workflow import ConcurrentBuilder
+
+    # Minimal: use default aggregator (returns list[ChatMessage])
+    workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).build()
+
+
+    # Custom aggregator via callback (sync or async). The callback receives
+    # list[AgentExecutorResponse] and its return value becomes
+    # WorkflowCompletedEvent.data
+    def summarize(results):
+        return " | ".join(r.agent_run_response.messages[-1].text for r in results)
+
+
+    workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_custom_aggregator(summarize).build()
+    ```
+    """
+
+    def __init__(self) -> None:
+        self._participants: list[AgentProtocol | Executor] = []
+        self._aggregator: Executor | None = None
+
+    def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "ConcurrentBuilder":
+        r"""Define the parallel participants for this concurrent workflow.
+
+        Accepts AgentProtocol instances (e.g., created by a chat client) or Executor
+        instances. Each participant is wired as a parallel branch using fan-out edges
+        from an internal dispatcher.
+
+        Raises:
+            ValueError: if `participants` is empty or contains duplicates
+            TypeError: if any entry is not AgentProtocol or Executor
+
+        Example:
+        ```python
+        wf = ConcurrentBuilder().participants([researcher_agent, marketer_agent, legal_agent]).build()
+
+        # Mixing agent(s) and executor(s) is supported
+        wf2 = ConcurrentBuilder().participants([researcher_agent, my_custom_executor]).build()
+        ```
+        """
+        if not participants:
+            raise ValueError("participants cannot be empty")
+
+        # Defensive duplicate detection
+        seen_agent_ids: set[int] = set()
+        seen_executor_ids: set[str] = set()
+        for p in participants:
+            if isinstance(p, Executor):
+                if p.id in seen_executor_ids:
+                    raise ValueError(f"Duplicate executor participant detected: id '{p.id}'")
+                seen_executor_ids.add(p.id)
+            elif isinstance(p, AgentProtocol):
+                pid = id(p)
+                if pid in seen_agent_ids:
+                    raise ValueError("Duplicate agent participant detected (same agent instance provided twice)")
+                seen_agent_ids.add(pid)
+            else:
+                raise TypeError(f"participants must be AgentProtocol or Executor instances; got {type(p).__name__}")
+
+        self._participants = list(participants)
+        return self
+
+    def with_aggregator(self, aggregator: Executor | Callable[..., Any]) -> "ConcurrentBuilder":
+        r"""Override the default aggregator with an Executor or a callback.
+
+        - Executor: must handle `list[AgentExecutorResponse]` and add a
+          `WorkflowCompletedEvent` to the context.
+        - Callback: sync or async callable with one of the signatures:
+          `(results: list[AgentExecutorResponse]) -> Any | None` or
+          `(results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None`.
+          If the callback returns a non-None value, it becomes the
+          `WorkflowCompletedEvent.data`.
+
+        Example:
+        ```python
+        # Callback-based aggregator (string result)
+        async def summarize(results):
+            return " | ".join(r.agent_run_response.messages[-1].text for r in results)
+
+
+        wf = ConcurrentBuilder().participants([a1, a2, a3]).with_custom_aggregator(summarize).build()
+        ```
+        """
+        if isinstance(aggregator, Executor):
+            self._aggregator = aggregator
+        elif callable(aggregator):
+            self._aggregator = _CallbackAggregator(aggregator)
+        else:
+            raise TypeError("aggregator must be an Executor or a callable")
+        return self
+
+    def build(self) -> Workflow:
+        r"""Build and validate the concurrent workflow.
+
+        Wiring pattern:
+        - Dispatcher (internal) fans out the input to all `participants`
+        - Fan-in aggregator collects `AgentExecutorResponse` objects
+        - Aggregator emits a `WorkflowCompletedEvent` with either:
+          - list[ChatMessage] (default aggregator: one user + one assistant per agent)
+          - custom payload from the provided callback/executor
+
+        Returns:
+            Workflow: a ready-to-run workflow instance
+
+        Raises:
+            ValueError: if no participants were defined
+
+        Example:
+        ```python
+        workflow = ConcurrentBuilder().participants([agent1, agent2]).build()
+        ```
+        """
+        if not self._participants:
+            raise ValueError("No participants provided. Call .participants([...]) first.")
+
+        dispatcher = _DispatchToAllParticipants(id="dispatcher")
+        aggregator = self._aggregator or _AggregateAgentConversations(id="aggregator")
+
+        builder = WorkflowBuilder()
+        return (
+            builder.set_start_executor(dispatcher)
+            .add_fan_out_edges(dispatcher, list(self._participants))
+            .add_fan_in_edges(list(self._participants), aggregator)
+            .build()
+        )

python/packages/workflow/tests/test_concurrent.py

@@ -0,0 +1,126 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import Any, cast
+
+import pytest
+from agent_framework import AgentRunResponse, ChatMessage, Role
+
+from agent_framework_workflow import (
+    AgentExecutorRequest,
+    AgentExecutorResponse,
+    ConcurrentBuilder,
+    Executor,
+    WorkflowCompletedEvent,
+    WorkflowContext,
+    handler,
+)
+
+
+class _FakeAgentExec(Executor):
+    """Test executor that mimics an agent by emitting an AgentExecutorResponse.
+
+    It takes the incoming AgentExecutorRequest, produces a single assistant message
+    with the configured reply text, and sends an AgentExecutorResponse that includes
+    full_conversation (the original user prompt followed by the assistant message).
+    """
+
+    def __init__(self, id: str, reply_text: str) -> None:
+        super().__init__(id)
+        self._reply_text = reply_text
+
+    @handler
+    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
+        response = AgentRunResponse(messages=ChatMessage(Role.ASSISTANT, text=self._reply_text))
+        full_conversation = list(request.messages) + list(response.messages)
+        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))
+
+
+def test_concurrent_builder_rejects_empty_participants() -> None:
+    with pytest.raises(ValueError):
+        ConcurrentBuilder().participants([])
+
+
+def test_concurrent_builder_rejects_duplicate_executors() -> None:
+    a = _FakeAgentExec("dup", "A")
+    b = _FakeAgentExec("dup", "B")  # same executor id
+    with pytest.raises(ValueError):
+        ConcurrentBuilder().participants([a, b])
+
+
+async def test_concurrent_default_aggregator_emits_single_user_and_assistants() -> None:
+    # Three synthetic agent executors
+    e1 = _FakeAgentExec("agentA", "Alpha")
+    e2 = _FakeAgentExec("agentB", "Beta")
+    e3 = _FakeAgentExec("agentC", "Gamma")
+
+    wf = ConcurrentBuilder().participants([e1, e2, e3]).build()
+
+    completed: WorkflowCompletedEvent | None = None
+    async for ev in wf.run_stream("prompt: hello world"):
+        if isinstance(ev, WorkflowCompletedEvent):
+            completed = ev
+            break
+
+    assert completed is not None
+    assert isinstance(completed.data, list)
+    messages: list[ChatMessage] = cast(list[ChatMessage], completed.data)  # type: ignore
+
+    # Expect one user message + one assistant message per participant
+    assert len(messages) == 1 + 3
+    assert messages[0].role == Role.USER
+    assert "hello world" in messages[0].text
+
+    assistant_texts = {m.text for m in messages[1:]}
+    assert assistant_texts == {"Alpha", "Beta", "Gamma"}
+    assert all(m.role == Role.ASSISTANT for m in messages[1:])
+
+
+async def test_concurrent_custom_aggregator_callback_is_used() -> None:
+    # Two synthetic agent executors for brevity
+    e1 = _FakeAgentExec("agentA", "One")
+    e2 = _FakeAgentExec("agentB", "Two")
+
+    async def summarize(results: list[AgentExecutorResponse]) -> str:
+        texts: list[str] = []
+        for r in results:
+            msgs: list[ChatMessage] = r.agent_run_response.messages
+            texts.append(msgs[-1].text if msgs else "")
+        return " | ".join(sorted(texts))
+
+    wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize).build()
+
+    completed: WorkflowCompletedEvent | None = None
+    async for ev in wf.run_stream("prompt: custom"):
+        if isinstance(ev, WorkflowCompletedEvent):
+            completed = ev
+            break
+
+    assert completed is not None
+    # Custom aggregator returns a string payload
+    assert isinstance(completed.data, str)
+    assert completed.data == "One | Two"
+
+
+async def test_concurrent_custom_aggregator_sync_callback_is_used() -> None:
+    e1 = _FakeAgentExec("agentA", "One")
+    e2 = _FakeAgentExec("agentB", "Two")
+
+    # Sync callback with ctx parameter (should run via asyncio.to_thread)
+    def summarize_sync(results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> str:  # type: ignore[unused-argument]
+        texts: list[str] = []
+        for r in results:
+            msgs: list[ChatMessage] = r.agent_run_response.messages
+            texts.append(msgs[-1].text if msgs else "")
+        return " | ".join(sorted(texts))
+
+    wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize_sync).build()
+
+    completed: WorkflowCompletedEvent | None = None
+    async for ev in wf.run_stream("prompt: custom sync"):
+        if isinstance(ev, WorkflowCompletedEvent):
+            completed = ev
+            break
+
+    assert completed is not None
+    assert isinstance(completed.data, str)
+    assert completed.data == "One | Two"