Add tests and more content types (#5235)

* Add tests

* fix tests and sample

* Fix formatting

* Remove function approval contents

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -1,7 +1,11 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
 import asyncio
-from collections.abc import AsyncIterable
+import json
+import logging
+from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping
 
 from agent_framework import ChatOptions, Content, HistoryProvider, Message, RawAgent, SupportsAgentRun
 from agent_framework._telemetry import append_to_user_agent
@@ -35,8 +39,18 @@ from azure.ai.agentserver.responses.models import (
     SummaryTextContent,
     TextContent,
 )
+from azure.ai.agentserver.responses.streaming._builders import (
+    OutputItemFunctionCallBuilder,
+    OutputItemMcpCallBuilder,
+    OutputItemMessageBuilder,
+    OutputItemReasoningItemBuilder,
+    ReasoningSummaryPartBuilder,
+    TextContentBuilder,
+)
 from typing_extensions import Any, Sequence, cast
 
+logger = logging.getLogger(__name__)
+
 
 class ResponsesHostServer(ResponsesAgentServerHost):
     """A responses server host for an agent."""
@@ -67,13 +81,18 @@ class ResponsesHostServer(ResponsesAgentServerHost):
         """
         super().__init__(prefix=prefix, options=options, store=store, **kwargs)
 
-        self._agent = agent
-        for provider in getattr(self._agent, "context_providers", []):
+        for provider in getattr(agent, "context_providers", []):
             if isinstance(provider, HistoryProvider) and provider.load_messages:
                 raise RuntimeError(
                     "There shouldn't be a history provider with `load_messages=True` already present. "
                     "History is managed by the hosting infrastructure."
                 )
+        self._agent = agent
+
+        self.create_handler(self._handle_create)  # pyright: ignore[reportUnknownMemberType]
+
+        # Append the user agent prefix for telemetry purposes
+        append_to_user_agent(self.USER_AGENT_PREFIX)
 
         self.create_handler(self._handle_create)  # pyright: ignore[reportUnknownMemberType]
 
@@ -98,8 +117,6 @@ class ResponsesHostServer(ResponsesAgentServerHost):
         yield stream.emit_created()
         yield stream.emit_in_progress()
 
-        # Add reasoning
-
         if request.stream is None or request.stream is False:
             # Run the agent in non-streaming mode
             if isinstance(self._agent, RawAgent):
@@ -107,35 +124,190 @@ class ResponsesHostServer(ResponsesAgentServerHost):
                 response = await raw_agent.run(messages, stream=False, options=chat_options)
             else:
                 response = await self._agent.run(messages, stream=False)
-            for item in stream.output_item_message(response.text):
-                yield item
+
+            for message in response.messages:
+                for content in message.contents:
+                    async for item in _to_outputs(stream, content):
+                        yield item
+
             yield stream.emit_completed()
             return
 
         # Start the streaming response
-        message_item = stream.add_output_item_message()
-        yield message_item.emit_added()
-        text_content = message_item.add_text_content()
-        yield text_content.emit_added()
-
-        # Invoke the MAF agent
         if isinstance(self._agent, RawAgent):
             raw_agent = cast("RawAgent[Any]", self._agent)  # pyright: ignore[reportUnknownMemberType]
             response_stream = raw_agent.run(messages, stream=True, options=chat_options)
         else:
             response_stream = self._agent.run(messages, stream=True)
-        async for update in response_stream:
-            if update.text:
-                yield text_content.emit_delta(update.text)
 
-        # Complete the message
-        yield text_content.emit_text_done()
-        yield text_content.emit_done()
-        yield message_item.emit_done()
+        # Track the current active output item builder for streaming;
+        # lazily created on matching content, closed when a different type arrives.
+        tracker = _OutputItemTracker(stream)
+
+        async for update in response_stream:
+            for content in update.contents:
+                for event in tracker.handle(content):
+                    yield event
+                if tracker.needs_async:
+                    async for item in _to_outputs(stream, content):
+                        yield item
+                    tracker.needs_async = False
+
+        # Close any remaining active builder
+        for event in tracker.close():
+            yield event
 
         yield stream.emit_completed()
 
 
+# region Active Builder State
+
+
+class _OutputItemTracker:
+    """Tracks the current active output item builder during streaming.
+
+    Handles lazy creation, delta emission, and closing of streaming builders
+    for text messages, reasoning, function calls, and MCP calls.
+    """
+
+    _DELTA_TYPES = frozenset({"text", "text_reasoning", "function_call", "mcp_server_tool_call"})
+
+    def __init__(self, stream: ResponseEventStream) -> None:
+        self._stream = stream
+        self._active_type: str | None = None
+        self._active_id: str | None = None
+        # Accumulated delta text for the current active builder
+        self._accumulated: list[str] = []
+        # Builder state — only one is active at a time
+        self._message_item: OutputItemMessageBuilder | None = None
+        self._text_content: TextContentBuilder | None = None
+        self._reasoning_item: OutputItemReasoningItemBuilder | None = None
+        self._summary_part: ReasoningSummaryPartBuilder | None = None
+        self._fc_builder: OutputItemFunctionCallBuilder | None = None
+        self._mcp_builder: OutputItemMcpCallBuilder | None = None
+        self.needs_async = False
+
+    def handle(self, content: Content) -> Generator[ResponseStreamEvent, None, None]:
+        """Process a content item, yielding sync events.
+
+        Sets ``needs_async = True`` if the caller must also drain an
+        async ``_to_outputs`` call for this content.
+        """
+        if content.type == "text" and content.text is not None:
+            if self._active_type != "text":
+                yield from self._close()
+                yield from self._open_message()
+            assert self._text_content is not None  # noqa: S101
+            self._accumulated.append(content.text)
+            yield self._text_content.emit_delta(content.text)
+
+        elif content.type == "text_reasoning" and content.text is not None:
+            if self._active_type != "text_reasoning":
+                yield from self._close()
+                yield from self._open_reasoning()
+            assert self._summary_part is not None  # noqa: S101
+            self._accumulated.append(content.text)
+            yield self._summary_part.emit_text_delta(content.text)
+
+        elif content.type == "function_call" and content.call_id is not None:
+            if self._active_type != "function_call" or self._active_id != content.call_id:
+                yield from self._close()
+                yield from self._open_function_call(content)
+            assert self._fc_builder is not None  # noqa: S101
+            args_str = _arguments_to_str(content.arguments)
+            self._accumulated.append(args_str)
+            yield self._fc_builder.emit_arguments_delta(args_str)
+
+        elif content.type == "mcp_server_tool_call" and content.tool_name:
+            key = f"{content.server_name or 'default'}::{content.tool_name}"
+            if self._active_type != "mcp_server_tool_call" or self._active_id != key:
+                yield from self._close()
+                yield from self._open_mcp_call(content)
+            assert self._mcp_builder is not None  # noqa: S101
+            args_str = _arguments_to_str(content.arguments)
+            self._accumulated.append(args_str)
+            yield self._mcp_builder.emit_arguments_delta(args_str)
+
+        else:
+            yield from self._close()
+            self.needs_async = True
+
+    def close(self) -> Generator[ResponseStreamEvent, None, None]:
+        """Close any remaining active builder."""
+        yield from self._close()
+
+    # -- Private open/close helpers --
+
+    def _open_message(self) -> Generator[ResponseStreamEvent, None, None]:
+        self._message_item = self._stream.add_output_item_message()
+        self._text_content = self._message_item.add_text_content()
+        self._active_type = "text"
+        self._active_id = None
+        yield self._message_item.emit_added()
+        yield self._text_content.emit_added()
+
+    def _open_reasoning(self) -> Generator[ResponseStreamEvent, None, None]:
+        self._reasoning_item = self._stream.add_output_item_reasoning_item()
+        self._summary_part = self._reasoning_item.add_summary_part()
+        self._active_type = "text_reasoning"
+        self._active_id = None
+        yield self._reasoning_item.emit_added()
+        yield self._summary_part.emit_added()
+
+    def _open_function_call(self, content: Content) -> Generator[ResponseStreamEvent, None, None]:
+        self._fc_builder = self._stream.add_output_item_function_call(
+            name=content.name or "",
+            call_id=content.call_id or "",
+        )
+        self._active_type = "function_call"
+        self._active_id = content.call_id
+        yield self._fc_builder.emit_added()
+
+    def _open_mcp_call(self, content: Content) -> Generator[ResponseStreamEvent, None, None]:
+        self._mcp_builder = self._stream.add_output_item_mcp_call(
+            server_label=content.server_name or "default",
+            name=content.tool_name or "",
+        )
+        self._active_type = "mcp_server_tool_call"
+        self._active_id = f"{content.server_name or 'default'}::{content.tool_name}"
+        yield self._mcp_builder.emit_added()
+
+    def _close(self) -> Generator[ResponseStreamEvent, None, None]:
+        accumulated = "".join(self._accumulated)
+
+        if self._active_type == "text" and self._text_content and self._message_item:
+            yield self._text_content.emit_text_done(accumulated)
+            yield self._text_content.emit_done()
+            yield self._message_item.emit_done()
+            self._text_content = None
+            self._message_item = None
+
+        elif self._active_type == "text_reasoning" and self._summary_part and self._reasoning_item:
+            yield self._summary_part.emit_text_done(accumulated)
+            yield self._summary_part.emit_done()
+            yield self._reasoning_item.emit_done()
+            self._summary_part = None
+            self._reasoning_item = None
+
+        elif self._active_type == "function_call" and self._fc_builder:
+            yield self._fc_builder.emit_arguments_done(accumulated)
+            yield self._fc_builder.emit_done()
+            self._fc_builder = None
+
+        elif self._active_type == "mcp_server_tool_call" and self._mcp_builder:
+            yield self._mcp_builder.emit_arguments_done(accumulated)
+            yield self._mcp_builder.emit_completed()
+            yield self._mcp_builder.emit_done()
+            self._mcp_builder = None
+
+        self._active_type = None
+        self._active_id = None
+        self._accumulated.clear()
+
+
+# endregion
+
+
 # region Option Conversion
 
 
@@ -165,7 +337,7 @@ def _to_chat_options(request: CreateResponse) -> ChatOptions:
 # endregion
 
 
-# region Message Conversion
+# region Input Message Conversion
 
 
 def _to_messages(history: Sequence[OutputItem]) -> list[Message]:
@@ -303,3 +475,110 @@ def _convert_message_content(content: MessageContent) -> Content:
 
 
 # endregion
+
+# region Output Item Conversion
+
+
+def _arguments_to_str(arguments: str | Mapping[str, Any] | None) -> str:
+    """Convert arguments to a JSON string.
+
+    Args:
+        arguments: The arguments to convert, can be a string, mapping, or None.
+
+    Returns:
+        The arguments as a JSON string.
+    """
+    if arguments is None:
+        return ""
+    if isinstance(arguments, str):
+        return arguments
+    return json.dumps(arguments)
+
+
+async def _to_outputs(stream: ResponseEventStream, content: Content) -> AsyncIterator[ResponseStreamEvent]:
+    """Converts a Content object to an async sequence of ResponseStreamEvent objects.
+
+    Args:
+        stream: The ResponseEventStream to use for building events.
+        content: The Content to convert.
+
+    Yields:
+        ResponseStreamEvent: The converted event objects.
+
+    Raises:
+        ValueError: If the Content type is not supported.
+    """
+    if content.type == "text" and content.text is not None:
+        async for event in stream.aoutput_item_message(content.text):
+            yield event
+    elif content.type == "text_reasoning" and content.text is not None:
+        async for event in stream.aoutput_item_reasoning_item(content.text):
+            yield event
+    elif content.type == "function_call":
+        async for event in stream.aoutput_item_function_call(
+            content.name,  # type: ignore[arg-type]
+            content.call_id,  # type: ignore[arg-type]
+            _arguments_to_str(content.arguments),
+        ):
+            yield event
+    elif content.type == "function_result":
+        async for event in stream.aoutput_item_function_call_output(
+            content.call_id,  # type: ignore[arg-type]
+            str(content.result or ""),
+        ):
+            yield event
+    elif content.type == "image_generation_tool_result" and content.outputs is not None:
+        async for event in stream.aoutput_item_image_gen_call(str(content.outputs)):
+            yield event
+    elif content.type == "mcp_server_tool_call":
+        mcp_call = stream.add_output_item_mcp_call(
+            server_label=content.server_name or "default",
+            name=content.tool_name or "",
+        )
+        yield mcp_call.emit_added()
+        async for event in mcp_call.aarguments(_arguments_to_str(content.arguments)):
+            yield event
+        yield mcp_call.emit_completed()
+        yield mcp_call.emit_done()
+    elif content.type == "mcp_server_tool_result":
+        output = (
+            content.output
+            if isinstance(content.output, str)
+            else str(content.output)
+            if content.output is not None
+            else ""
+        )
+        async for event in stream.aoutput_item_custom_tool_call_output(content.call_id or "", output):
+            yield event
+    elif content.type == "shell_tool_call":
+        action: dict[str, Any] = {"type": "exec", "command": content.commands or []}
+        async for event in stream.aoutput_item_function_shell_call(
+            content.call_id or "",
+            action,
+            {},
+            status=content.status or "completed",
+        ):
+            yield event
+    elif content.type == "shell_tool_result":
+        output_items: list[dict[str, Any]] = []
+        if content.outputs:
+            for out in content.outputs:
+                output_items.append({
+                    "type": "shell_output",
+                    "stdout": getattr(out, "stdout", "") or "",
+                    "stderr": getattr(out, "stderr", "") or "",
+                    "exit_code": getattr(out, "exit_code", None),
+                })
+        async for event in stream.aoutput_item_function_shell_call_output(
+            content.call_id or "",
+            output_items,
+            status=content.status or "completed",
+            max_output_length=content.max_output_length,
+        ):
+            yield event
+    else:
+        # Log a warning for unsupported content types instead of raising an error to avoid breaking the response stream.
+        logger.warning(f"Content type '{content.type}' is not supported yet.")
+
+
+# endregion

python/packages/foundry_hosting/tests/test_responses.py

@@ -0,0 +1,524 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""HTTP round-trip tests for ResponsesHostServer.
+
+These tests exercise the full HTTP pipeline using httpx.AsyncClient with
+ASGITransport — no real server process is started. Requests go through
+the Starlette routing stack, the Responses API middleware, and arrive at
+the registered _handle_create handler.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+from agent_framework import (
+    AgentResponse,
+    AgentResponseUpdate,
+    Content,
+    HistoryProvider,
+    Message,
+    RawAgent,
+    ResponseStream,
+)
+from azure.ai.agentserver.responses import InMemoryResponseProvider
+from typing_extensions import Any
+
+from agent_framework_foundry_hosting import ResponsesHostServer
+
+# region Helpers
+
+
+def _make_agent(
+    *,
+    response: AgentResponse | None = None,
+    stream_updates: list[AgentResponseUpdate] | None = None,
+) -> MagicMock:
+    """Create a mock agent implementing SupportsAgentRun."""
+    agent = MagicMock(spec=RawAgent)
+    agent.id = "test-agent"
+    agent.name = "Test Agent"
+    agent.description = "A mock agent for testing"
+    agent.context_providers = []
+
+    if response is not None:
+
+        async def run_non_streaming(*args: Any, **kwargs: Any) -> AgentResponse:
+            return response
+
+        agent.run = AsyncMock(side_effect=run_non_streaming)
+
+    if stream_updates is not None:
+
+        async def _stream_gen() -> AsyncIterator[AgentResponseUpdate]:
+            for update in stream_updates:
+                yield update
+
+        def run_streaming(*args: Any, **kwargs: Any) -> Any:
+            if kwargs.get("stream"):
+                return ResponseStream(_stream_gen())  # type: ignore
+            raise NotImplementedError("Only streaming is configured on this mock")
+
+        agent.run = MagicMock(side_effect=run_streaming)
+
+    return agent
+
+
+def _make_server(agent: MagicMock, **kwargs: Any) -> ResponsesHostServer:
+    """Create a ResponsesHostServer with an in-memory store."""
+    return ResponsesHostServer(agent, store=InMemoryResponseProvider(), **kwargs)
+
+
+async def _post(
+    server: ResponsesHostServer,
+    *,
+    input_text: str = "Hello",
+    model: str = "test-model",
+    stream: bool = False,
+    temperature: float | None = None,
+    top_p: float | None = None,
+    max_output_tokens: int | None = None,
+    parallel_tool_calls: bool | None = None,
+) -> httpx.Response:
+    """Send a POST /responses request through the ASGI transport."""
+    payload: dict[str, Any] = {"model": model, "input": input_text, "stream": stream}
+    if temperature is not None:
+        payload["temperature"] = temperature
+    if top_p is not None:
+        payload["top_p"] = top_p
+    if max_output_tokens is not None:
+        payload["max_output_tokens"] = max_output_tokens
+    if parallel_tool_calls is not None:
+        payload["parallel_tool_calls"] = parallel_tool_calls
+
+    transport = httpx.ASGITransport(app=server)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        return await client.post("/responses", json=payload)
+
+
+def _parse_sse_events(body: str) -> list[dict[str, Any]]:
+    """Parse SSE text into a list of event dicts with 'event' and 'data' keys."""
+    events: list[dict[str, Any]] = []
+    current_event: str | None = None
+    current_data_lines: list[str] = []
+
+    for line in body.split("\n"):
+        if line.startswith("event: "):
+            current_event = line[len("event: ") :]
+        elif line.startswith("data: "):
+            current_data_lines.append(line[len("data: ") :])
+        elif line.strip() == "" and current_event is not None:
+            data_str = "\n".join(current_data_lines)
+            try:
+                data = json.loads(data_str)
+            except json.JSONDecodeError:
+                data = data_str
+            events.append({"event": current_event, "data": data})
+            current_event = None
+            current_data_lines = []
+
+    return events
+
+
+def _sse_event_types(events: list[dict[str, Any]]) -> list[str]:
+    """Extract event type strings from parsed SSE events."""
+    return [e["event"] for e in events]
+
+
+# endregion
+
+
+# region Initialization
+
+
+class TestResponsesHostServerInit:
+    def test_init_basic(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        server = _make_server(agent)
+        assert server is not None
+
+    def test_init_rejects_history_provider_with_load_messages(self) -> None:
+        hp = HistoryProvider(source_id="test", load_messages=True)
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        agent.context_providers = [hp]
+        with pytest.raises(RuntimeError, match="history provider"):
+            ResponsesHostServer(agent)
+
+
+# endregion
+
+
+# region Health Check
+
+
+class TestHealthCheck:
+    async def test_readiness(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        server = _make_server(agent)
+        transport = httpx.ASGITransport(app=server)
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.get("/readiness")
+        assert resp.status_code == 200
+
+
+# endregion
+
+
+# region Non-streaming
+
+
+class TestNonStreaming:
+    async def test_basic_text_response(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("Hello!")])])
+        )
+        server = _make_server(agent)
+        resp = await _post(server, input_text="Hi", stream=False)
+
+        assert resp.status_code == 200
+        assert "application/json" in resp.headers["content-type"]
+
+        body = resp.json()
+        assert body["object"] == "response"
+        assert body["status"] == "completed"
+        assert len(body["output"]) > 0
+
+        # Find the message output item with our text
+        text_found = False
+        for item in body["output"]:
+            assert item["type"] == "message"
+            for part in item.get("content", []):
+                if part.get("type") == "output_text" and part.get("text") == "Hello!":
+                    text_found = True
+        assert text_found, f"Expected 'Hello!' in output, got: {body['output']}"
+
+    async def test_function_call_and_result(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(
+                messages=[
+                    Message(
+                        role="assistant",
+                        contents=[Content.from_function_call("call_1", "get_weather", arguments='{"loc": "NYC"}')],
+                    ),
+                    Message(role="tool", contents=[Content.from_function_result("call_1", result="sunny")]),
+                    Message(role="assistant", contents=[Content.from_text("The weather is sunny!")]),
+                ]
+            )
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=False)
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "completed"
+
+        types = [item["type"] for item in body["output"]]
+        assert "function_call" in types
+        assert "function_call_output" in types
+        assert "message" in types
+
+    async def test_reasoning_content(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(
+                messages=[
+                    Message(
+                        role="assistant",
+                        contents=[
+                            Content.from_text_reasoning(text="Let me think..."),
+                            Content.from_text("The answer is 42"),
+                        ],
+                    ),
+                ]
+            )
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=False)
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "completed"
+
+        types = [item["type"] for item in body["output"]]
+        assert "reasoning" in types
+        assert "message" in types
+
+    async def test_empty_response(self) -> None:
+        agent = _make_agent(response=AgentResponse(messages=[]))
+        server = _make_server(agent)
+        resp = await _post(server, stream=False)
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "completed"
+
+    async def test_chat_options_forwarded(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=False, temperature=0.5, top_p=0.9, max_output_tokens=1024)
+
+        assert resp.status_code == 200
+        agent.run.assert_awaited_once()
+        call_kwargs = agent.run.call_args.kwargs
+        assert call_kwargs["stream"] is False
+        options = call_kwargs["options"]
+        assert options["temperature"] == 0.5
+        assert options["top_p"] == 0.9
+        assert options["max_tokens"] == 1024
+
+
+# endregion
+
+
+# region Streaming
+
+
+class TestStreaming:
+    async def test_basic_text_streaming(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(contents=[Content.from_text("Hello ")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("world!")], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        assert "text/event-stream" in resp.headers["content-type"]
+
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[0] == "response.created"
+        assert types[1] == "response.in_progress"
+        assert types[-1] == "response.completed"
+        assert "response.output_text.delta" in types
+        assert types.count("response.output_text.delta") == 2
+        assert "response.output_text.done" in types
+
+        # Verify the accumulated text in the done event
+        done_events = [e for e in events if e["event"] == "response.output_text.done"]
+        assert len(done_events) == 1
+        assert done_events[0]["data"]["text"] == "Hello world!"
+
+    async def test_function_call_streaming(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments='{"q":')],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments=' "hello"}')],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[0] == "response.created"
+        assert types[-1] == "response.completed"
+        assert types.count("response.function_call_arguments.delta") == 2
+        assert "response.function_call_arguments.done" in types
+
+        # Verify accumulated arguments
+        args_done = [e for e in events if e["event"] == "response.function_call_arguments.done"]
+        assert len(args_done) == 1
+        assert args_done[0]["data"]["arguments"] == '{"q": "hello"}'
+
+    async def test_alternating_text_and_function_call(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                # Text deltas
+                AgentResponseUpdate(contents=[Content.from_text("Let me ")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("search...")], role="assistant"),
+                # Function call argument deltas
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments='{"q":')],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments=' "x"}')],
+                    role="assistant",
+                ),
+                # More text deltas
+                AgentResponseUpdate(contents=[Content.from_text("Found ")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("it!")], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[0] == "response.created"
+        assert types[-1] == "response.completed"
+
+        # 4 text deltas + 2 function call argument deltas
+        assert types.count("response.output_text.delta") == 4
+        assert types.count("response.function_call_arguments.delta") == 2
+
+        # 3 distinct output items (text, fc, text)
+        assert types.count("response.output_item.added") == 3
+        assert types.count("response.output_item.done") == 3
+
+        # Verify accumulated content
+        text_done = [e for e in events if e["event"] == "response.output_text.done"]
+        assert len(text_done) == 2
+        assert text_done[0]["data"]["text"] == "Let me search..."
+        assert text_done[1]["data"]["text"] == "Found it!"
+
+        args_done = [e for e in events if e["event"] == "response.function_call_arguments.done"]
+        assert len(args_done) == 1
+        assert args_done[0]["data"]["arguments"] == '{"q": "x"}'
+
+    async def test_reasoning_then_text_streaming(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                # Reasoning deltas
+                AgentResponseUpdate(contents=[Content.from_text_reasoning(text="Let me ")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text_reasoning(text="think...")], role="assistant"),
+                # Text deltas
+                AgentResponseUpdate(contents=[Content.from_text("The answer ")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("is 42")], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[0] == "response.created"
+        assert types[-1] == "response.completed"
+        # Reasoning + text = 2 output items
+        assert types.count("response.output_item.added") == 2
+        assert types.count("response.output_item.done") == 2
+        assert types.count("response.output_text.delta") == 2
+
+        # Verify accumulated text
+        text_done = [e for e in events if e["event"] == "response.output_text.done"]
+        assert len(text_done) == 1
+        assert text_done[0]["data"]["text"] == "The answer is 42"
+
+    async def test_empty_streaming(self) -> None:
+        agent = _make_agent(stream_updates=[])
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types == ["response.created", "response.in_progress", "response.completed"]
+
+    async def test_mixed_contents_in_single_update(self) -> None:
+        """Text and function call in one update switches builder mid-update."""
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_text("Let me search"),
+                        Content.from_function_call("call_1", "search", arguments='{"q": "test"}'),
+                    ],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert "response.output_text.delta" in types
+        assert "response.output_text.done" in types
+        assert "response.function_call_arguments.delta" in types
+        assert "response.function_call_arguments.done" in types
+
+    async def test_different_function_call_ids_produce_separate_items(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "func_a", arguments='{"x":1}')],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_2", "func_b", arguments='{"y":2}')],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        # Two separate function call items
+        assert types.count("response.output_item.added") == 2
+        assert types.count("response.function_call_arguments.done") == 2
+
+    async def test_mcp_tool_call_streaming(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[
+                        Content(
+                            type="mcp_server_tool_call",
+                            server_name="my_server",
+                            tool_name="search",
+                            arguments='{"query":',
+                        )
+                    ],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[
+                        Content(
+                            type="mcp_server_tool_call",
+                            server_name="my_server",
+                            tool_name="search",
+                            arguments=' "test"}',
+                        )
+                    ],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[0] == "response.created"
+        assert types[-1] == "response.completed"
+        assert "response.output_item.added" in types
+        assert "response.output_item.done" in types
+
+
+# endregion