Add MCP builder support for output/error and caller-supplied item IDs

sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/streaming/_builders/_tools.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterable
-from typing import TYPE_CHECKING, AsyncIterator, Iterator, cast
+from typing import TYPE_CHECKING, Any, AsyncIterator, Iterator, cast
 
 from ...models import _generated as generated_models
 from ._base import BaseOutputItemBuilder, _require_non_empty
@@ -540,26 +540,39 @@ def emit_failed(self) -> generated_models.ResponseMCPCallFailedEvent:
             self._emit_item_state_event(generated_models.ResponseStreamEventType.RESPONSE_MCP_CALL_FAILED.value),
         )
 
-    def emit_done(self) -> generated_models.ResponseOutputItemDoneEvent:
+    def emit_done(
+        self,
+        *,
+        output: str | None = None,
+        error: dict[str, Any] | None = None,
+    ) -> generated_models.ResponseOutputItemDoneEvent:
         """Emit an ``output_item.done`` event for this MCP call.
 
         The ``status`` field reflects the most recent terminal state event
         (``emit_completed`` or ``emit_failed``). Defaults to ``"completed"``
         if neither was called.
 
+        :keyword output: Optional MCP tool output payload.
+        :keyword type output: str | None
+        :keyword error: Optional MCP tool error payload.
+        :keyword type error: dict[str, Any] | None
+
         :returns: The emitted event dict.
         :rtype: ResponseOutputItemDoneEvent
         """
-        return self._emit_done(
-            {
-                "type": "mcp_call",
-                "id": self._item_id,
-                "server_label": self._server_label,
-                "name": self._name,
-                "arguments": self._final_arguments or "",
-                "status": self._terminal_status or "completed",
-            }
-        )
+        item: dict[str, Any] = {
+            "type": "mcp_call",
+            "id": self._item_id,
+            "server_label": self._server_label,
+            "name": self._name,
+            "arguments": self._final_arguments or "",
+            "status": self._terminal_status or "completed",
+        }
+        if output is not None:
+            item["output"] = output
+        if error is not None:
+            item["error"] = error
+        return self._emit_done(item)
 
     # ---- Sub-item convenience generators (S-053) ----
 

sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/streaming/_event_stream.py

@@ -443,23 +443,38 @@ def add_output_item_image_gen_call(self) -> OutputItemImageGenCallBuilder:
         item_id = IdGenerator.new_image_gen_call_item_id(self._response_id)
         return OutputItemImageGenCallBuilder(self, output_index=output_index, item_id=item_id)
 
-    def add_output_item_mcp_call(self, server_label: str, name: str) -> OutputItemMcpCallBuilder:
+    def add_output_item_mcp_call(
+        self,
+        server_label: str,
+        name: str,
+        *,
+        item_id: str | None = None,
+    ) -> OutputItemMcpCallBuilder:
         """Add an MCP tool call output item and return its scoped builder.
 
         :param server_label: Label identifying the MCP server.
         :type server_label: str
         :param name: Name of the MCP tool being called.
         :type name: str
+        :keyword item_id: Optional caller-supplied output item identifier.
+        :keyword type item_id: str | None
         :returns: A builder for emitting MCP call argument deltas and lifecycle events.
         :rtype: OutputItemMcpCallBuilder
         """
         output_index = self._output_index
         self._output_index += 1
-        item_id = IdGenerator.new_mcp_call_item_id(self._response_id)
+        if item_id is None:
+            resolved_item_id = IdGenerator.new_mcp_call_item_id(self._response_id)
+        else:
+            if not isinstance(item_id, str):
+                raise TypeError("item_id must be a string")
+            resolved_item_id = item_id.strip()
+            if not resolved_item_id:
+                raise ValueError("item_id must be a non-empty string")
         return OutputItemMcpCallBuilder(
             self,
             output_index=output_index,
-            item_id=item_id,
+            item_id=resolved_item_id,
             server_label=server_label,
             name=name,
         )

sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_builders.py

@@ -278,6 +278,31 @@ def test_stream_item_id_generation__uses_expected_shape_and_response_partition_k
         assert len(body) == 50
 
 
+def test_add_output_item_mcp_call__uses_caller_supplied_item_id() -> None:
+    stream = ResponseEventStream(response_id=IdGenerator.new_response_id())
+    stream.emit_created()
+
+    mcp_call = stream.add_output_item_mcp_call("srv", "tool", item_id="mcp_06b686e11f")
+
+    assert mcp_call.item_id == "mcp_06b686e11f"
+
+
+def test_output_item_mcp_call_emit_done__includes_output_and_error_when_provided() -> None:
+    stream = ResponseEventStream(response_id=IdGenerator.new_response_id())
+    stream.emit_created()
+
+    mcp_call = stream.add_output_item_mcp_call("srv", "tool", item_id="mcp_custom")
+    mcp_call.emit_added()
+    mcp_call.emit_arguments_done('{"arg": 1}')
+    mcp_call.emit_failed()
+    done = mcp_call.emit_done(output='{"value": 42}', error={"code": "tool_error"})
+
+    assert done["type"] == "response.output_item.done"
+    assert done["item"]["id"] == "mcp_custom"
+    assert done["item"]["output"] == '{"value": 42}'
+    assert done["item"]["error"] == {"code": "tool_error"}
+
+
 def test_response_event_stream__exposes_mutable_response_snapshot_for_lifecycle_events() -> None:
     stream = ResponseEventStream(response_id="resp_builder_snapshot", model="gpt-4o-mini")
     stream.response.temperature = 1

sdk/agentserver/azure-ai-agentserver-responses/tests/unit/test_emit_return_types.py

@@ -787,6 +787,15 @@ def test_emit_done(self) -> None:
         event = mcp.emit_done()
         assert isinstance(event, ResponseOutputItemDoneEvent)
 
+    def test_emit_done_with_output_and_error(self) -> None:
+        s = _stream()
+        s.emit_created()
+        mcp = s.add_output_item_mcp_call("server", "tool", item_id="mcp_test")
+        mcp.emit_added()
+        mcp.emit_failed()
+        event = mcp.emit_done(output="ok", error={"reason": "failed"})
+        assert isinstance(event, ResponseOutputItemDoneEvent)
+
 
 # =====================================================================
 # OutputItemMcpListToolsBuilder