diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 567335d2f..0e404e869 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -3,15 +3,17 @@ from __future__ import annotations import asyncio +import base64 import json import logging import os import tempfile import threading -from collections.abc import AsyncIterable, AsyncIterator, Generator +from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping, Sequence from contextlib import AbstractAsyncContextManager, AsyncExitStack, suppress +from dataclasses import asdict, dataclass, is_dataclass from pathlib import Path -from typing import cast +from typing import Protocol, cast from agent_framework import ( ChatOptions, @@ -19,6 +21,7 @@ from agent_framework import ( ContextProvider, FileCheckpointStorage, HistoryProvider, + Message, RawAgent, SupportsAgentRun, WorkflowAgent, @@ -29,10 +32,78 @@ from azure.ai.agentserver.responses import ( ResponseEventStream, ResponseProviderProtocol, ResponsesServerOptions, - models, ) from azure.ai.agentserver.responses._id_generator import IdGenerator from azure.ai.agentserver.responses.hosting import ResponsesAgentServerHost +from azure.ai.agentserver.responses.models import ( + ApplyPatchToolCallItemParam, + ApplyPatchToolCallOutputItemParam, + ComputerCallOutputItemParam, + ComputerScreenshotContent, + CreateResponse, + FunctionCallOutputItemParam, + FunctionShellAction, + FunctionShellCallItemParam, + FunctionShellCallOutputContent, + FunctionShellCallOutputExitOutcome, + FunctionShellCallOutputItemParam, + Item, + ItemCodeInterpreterToolCall, + ItemComputerToolCall, + ItemCustomToolCall, + ItemCustomToolCallOutput, + ItemFileSearchToolCall, + ItemFunctionToolCall, + ItemImageGenToolCall, + ItemLocalShellToolCall, + ItemLocalShellToolCallOutput, + ItemMcpApprovalRequest, + ItemMcpToolCall, + ItemMessage, + ItemOutputMessage, + ItemReasoningItem, + ItemWebSearchToolCall, + LocalEnvironmentResource, + MCPApprovalResponse, + MessageContent, + MessageContentInputFileContent, + MessageContentInputImageContent, + MessageContentInputTextContent, + MessageContentOutputTextContent, + MessageContentReasoningTextContent, + MessageContentRefusalContent, + MessageRole, + OAuthConsentRequestOutputItem, + OutputItem, + OutputItemApplyPatchToolCall, + OutputItemApplyPatchToolCallOutput, + OutputItemCodeInterpreterToolCall, + OutputItemComputerToolCall, + OutputItemComputerToolCallOutputResource, + OutputItemCustomToolCall, + OutputItemCustomToolCallOutput, + OutputItemFileSearchToolCall, + OutputItemFunctionShellCall, + OutputItemFunctionShellCallOutput, + OutputItemFunctionToolCall, + OutputItemImageGenToolCall, + OutputItemLocalShellToolCall, + OutputItemLocalShellToolCallOutput, + OutputItemMcpApprovalRequest, + OutputItemMcpApprovalResponseResource, + OutputItemMcpToolCall, + OutputItemMessage, + OutputItemOutputMessage, + OutputItemReasoningItem, + OutputItemWebSearchToolCall, + OutputMessageContent, + OutputMessageContentOutputTextContent, + OutputMessageContentRefusalContent, + ResponseStreamEvent, + StructuredOutputsOutputItem, + SummaryTextContent, + TextContent, +) from azure.ai.agentserver.responses.streaming._builders import ( OutputItemFunctionCallBuilder, OutputItemMcpCallBuilder, @@ -44,44 +115,23 @@ from azure.ai.agentserver.responses.streaming._builders import ( from mcp import McpError from typing_extensions import Any -from ._shared import ( - ApprovalStorage, - _arguments_to_str, # pyright: ignore[reportPrivateUsage] - _convert_message_content, # pyright: ignore[reportPrivateUsage] - _convert_output_message_content, # pyright: ignore[reportPrivateUsage] - _item_to_message, # pyright: ignore[reportPrivateUsage] - _items_to_messages, # pyright: ignore[reportPrivateUsage] - _output_item_to_message, # pyright: ignore[reportPrivateUsage] - _output_items_to_messages, # pyright: ignore[reportPrivateUsage] -) - -# Re-export the conversion helpers under their historical names so existing -# tests (which import them from this module) keep working — the canonical -# definitions now live in :mod:`._shared`. -__all__ = ( - "ApprovalStorage", - "_arguments_to_str", - "_convert_message_content", - "_convert_output_message_content", - "_item_to_message", - "_items_to_messages", - "_output_item_to_message", - "_output_items_to_messages", -) - -# Local aliases for the agent-server SDK types this module touches at the -# Python type-annotation layer. Using ``models.X`` everywhere would work but -# would noisily clutter type-only positions where the alias adds no value. -CreateResponse = models.CreateResponse -ResponseStreamEvent = models.ResponseStreamEvent -FunctionShellAction = models.FunctionShellAction -FunctionShellCallOutputContent = models.FunctionShellCallOutputContent -FunctionShellCallOutputExitOutcome = models.FunctionShellCallOutputExitOutcome -LocalEnvironmentResource = models.LocalEnvironmentResource -OAuthConsentRequestOutputItem = models.OAuthConsentRequestOutputItem - logger = logging.getLogger(__name__) +_AZURE_RESPONSES_MESSAGE_ROLE_TYPE = f"{MessageRole.__module__}:{MessageRole.__qualname__}" + + +# region Approval Storage +class ApprovalStorage(Protocol): + """Storage for saving function approval requests.""" + + async def save_approval_request(self, approval_request_id: str, request: Content) -> None: + """Save a function approval request under the given ID.""" + ... + + async def load_approval_request(self, approval_request_id: str) -> Content: + """Load a function approval request by its ID.""" + ... + class InMemoryFunctionApprovalStorage: """An in-memory storage for function approval requests.""" @@ -202,35 +252,85 @@ def _checkpoint_storage_for_context(root: str, context_id: str) -> FileCheckpoin storage_path = (root_path / context_id).resolve() if not storage_path.is_relative_to(root_path): raise RuntimeError(f"Invalid checkpoint context id: {context_id!r}") - return FileCheckpointStorage(storage_path) + return FileCheckpointStorage( + storage_path, + # Keep this provider-specific allowlist narrow. Hosted workflow + # checkpoints can persist Azure's role enum inside Message objects. + allowed_checkpoint_types=[_AZURE_RESPONSES_MESSAGE_ROLE_TYPE], + ) # endregion Approval Storage # Foundry Toolbox Auth integration # Consent-URL error code returned by the Foundry MCP gateway when calling `/list` -CONSENT_ERROR_CODE = -32007 +CONSENT_ERROR_CODE = -32006 -def consent_url_from_error(exc: BaseException) -> str | None: - """Return the consent URL when ``exc`` wraps a Foundry MCP gateway consent error. +@dataclass +class ConsentError: + name: str + consent_url: str - The Agent Framework MCP layer surfaces gateway consent failures by wrapping the underlying - ``McpError`` inside an :class:`AgentFrameworkException` (typically a ``ToolExecutionException`` - raised from ``MCPStreamableHTTPTool.__aenter__``). This helper inspects ``exc.args`` for a - wrapped ``McpError`` whose ``error.code`` is :data:`CONSENT_ERROR_CODE`; when found, the - consent link the gateway returned in ``error.message`` is returned. Returns ``None`` for - anything else, so callers can do ``if (url := consent_url_from_error(ex)) is None: raise``. + +def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None: + """Return the consent URLs when ``exc`` wraps Foundry MCP gateway consent errors. Args: exc: The exception to inspect. Returns: - The consent URL if ``exc`` wraps a consent ``McpError``, otherwise ``None``. + The consent URL(s) extracted from the error, or ``None`` if no consent error was found. """ inner_exception = next((arg for arg in exc.args if isinstance(arg, McpError)), None) if inner_exception is not None and inner_exception.error.code == CONSENT_ERROR_CODE: - return inner_exception.error.message + # Parse the error message + # The error message is structured with the following format: + # "tools/list failed for 1 tool source(s), succeeded for 0 tool source(s) {"errors":[{"name": ..." + # where the second part is a JSON string that can be deserialized into an object with the following shape: + # ruff: disable[ERA001] + # { + # "errors" : [ + # { + # "name": "Name of the MCP tool that requires consent", + # "type" : "mcp", + # "error": { + # "code": "CONSENT_REQUIRED", + # "message": consent_url, + # } + # } + # ] + # } + # ruff: enable[ERA001] + try: + consent_errors: list[ConsentError] = [] + error_message_start = inner_exception.error.message.find("{") + if error_message_start == -1: + logger.warning("Consent error message does not contain JSON: %s", inner_exception.error.message) + return None + consent_details_json = inner_exception.error.message[error_message_start:] + consent_details = json.loads(consent_details_json) + if "errors" not in consent_details or not isinstance(consent_details["errors"], list): + logger.warning("Consent error message JSON does not contain 'errors' list: %s", consent_details_json) + return None + for error in consent_details["errors"]: + if ( + isinstance(error, dict) + and error.get("type") == "mcp" # type: ignore + and "error" in error + and isinstance(error["error"], dict) + and error["error"].get("code") == "CONSENT_REQUIRED" # type: ignore + and "message" in error["error"] + ): + consent_url = error["error"]["message"] # type: ignore + if isinstance(consent_url, str): + consent_errors.append(ConsentError(name=error.get("name", "Unknown"), consent_url=consent_url)) # type: ignore + else: + logger.warning("Consent URL in error message is not a valid URL: %s", consent_url) # type: ignore + if consent_errors: + return consent_errors + except json.JSONDecodeError: + logger.warning("Failed to parse consent details JSON: %s", inner_exception.error.message) return None @@ -361,71 +461,70 @@ class ResponsesHostServer(ResponsesAgentServerHost): context: ResponseContext, ) -> AsyncIterable[ResponseStreamEvent | dict[str, Any]]: """Handle the creation of a response for a regular (non-workflow) agent.""" - input_items = await context.get_input_items() - input_messages = await _items_to_messages(input_items, approval_storage=self._approval_storage) - - history = await context.get_history() - run_kwargs: dict[str, Any] = { - "messages": [ - *(await _output_items_to_messages(history, approval_storage=self._approval_storage)), - *input_messages, - ] - } - is_streaming_request = request.stream is not None and request.stream is True - - chat_options, are_options_set = _to_chat_options(request) - response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model) - yield response_event_stream.emit_created() yield response_event_stream.emit_in_progress() - if are_options_set and not isinstance(self._agent, RawAgent): - logger.warning("Agent doesn't support runtime options. They will be ignored.") - else: - run_kwargs["options"] = chat_options - - # Lazy-enter the agent (and any MCP tools it owns). The MCP client wraps gateway - # consent failures (and other connection-time errors) in AgentFrameworkException; if - # one of those is a consent error we surface the consent link to the client through - # the already-opened response stream instead of crashing the request. Other exception - # types propagate normally so the host can handle / log them. - try: - await self._ensure_agent_ready() - except AgentFrameworkException as ex: - consent_url = consent_url_from_error(ex) - if consent_url is None: - raise - logger.warning("OAuth consent required for Foundry MCP gateway.") - oauth_item = OAuthConsentRequestOutputItem( - id=IdGenerator.new_id("oacr"), - consent_link=consent_url, - server_label="Foundry Toolbox", - ) - builder = response_event_stream.add_output_item(oauth_item.id) - yield builder.emit_added(oauth_item) - yield builder.emit_done(oauth_item) - yield response_event_stream.emit_completed() - return - # Track the current active output item builder for streaming; # lazily created on matching content, closed when a different type arrives. - tracker: _OutputItemTracker | None = _OutputItemTracker(response_event_stream) if is_streaming_request else None + tracker: _OutputItemTracker | None = None try: + input_items = await context.get_input_items() + input_messages = await _items_to_messages(input_items, approval_storage=self._approval_storage) + + history = await context.get_history() + run_kwargs: dict[str, Any] = { + "messages": [ + *(await _output_items_to_messages(history, approval_storage=self._approval_storage)), + *input_messages, + ] + } + is_streaming_request = request.stream is not None and request.stream is True + + chat_options, are_options_set = _to_chat_options(request) + + if are_options_set and not isinstance(self._agent, RawAgent): + logger.warning("Agent doesn't support runtime options. They will be ignored.") + else: + run_kwargs["options"] = chat_options + + # Lazy-enter the agent (and any MCP tools it owns). The MCP client wraps gateway + # consent failures (and other connection-time errors) in AgentFrameworkException; if + # one of those is a consent error we surface the consent link to the client through + # the already-opened response stream instead of failing the request. Other exception + # types fall through to the outer handler below and become ``response.failed``. + try: + await self._ensure_agent_ready() + except AgentFrameworkException as ex: + consent_errors = consent_url_from_error(ex) + if consent_errors is None: + raise + for consent_error in consent_errors: + logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_error.consent_url) + oauth_item = OAuthConsentRequestOutputItem( + id=IdGenerator.new_id("oacr"), + consent_link=consent_error.consent_url, + server_label=consent_error.name, + ) + builder = response_event_stream.add_output_item(oauth_item.id) + yield builder.emit_added(oauth_item) + yield builder.emit_done(oauth_item) + yield response_event_stream.emit_completed() + return + + tracker = _OutputItemTracker(response_event_stream) if is_streaming_request else None + if not is_streaming_request: # Run the agent in non-streaming mode response = await self._agent.run(stream=False, **run_kwargs) # type: ignore[reportUnknownMemberType] - for message in response.messages: - for content in message.contents: - async for item in _to_outputs( - response_event_stream, - content, - approval_storage=self._approval_storage, - ): - yield item - yield response_event_stream.emit_completed() + async for item in _to_outputs_for_messages( + response_event_stream, + response.messages, + approval_storage=self._approval_storage, + ): + yield item else: if tracker is None: # pragma: no cover - defensive, set above raise RuntimeError("Streaming tracker was not initialized.") @@ -446,160 +545,158 @@ class ResponsesHostServer(ResponsesAgentServerHost): # Close any remaining active builder for event in tracker.close(): yield event - yield response_event_stream.emit_completed() - except Exception: - # Drain any in-progress streaming builder before emitting consent - # so the resulting stream stays well-formed. - if tracker is not None: - for event in tracker.close(): - yield event - yield response_event_stream.emit_completed() - raise + yield response_event_stream.emit_completed() + except Exception as ex: + logger.exception("Failed to produce response for agent") + for event in self._emit_failure(response_event_stream, tracker, ex): + yield event async def _handle_inner_workflow( self, request: CreateResponse, context: ResponseContext, ) -> AsyncIterable[ResponseStreamEvent | dict[str, Any]]: - """Handle the creation of a response for a workflow agent. - - Why this is required: - The sandbox may be deactivated after some period of inactivity, and only data managed - by the hosting infrastructure or files will be preserved upon deactivation. - """ - input_items = await context.get_input_items() - input_messages = await _items_to_messages(input_items, approval_storage=self._approval_storage) - is_streaming_request = request.stream is not None and request.stream is True - - _, are_options_set = _to_chat_options(request) - if are_options_set: - logger.warning("Workflow agent doesn't support runtime options. They will be ignored.") - - if request.previous_response_id is not None and context.conversation_id is not None: - raise RuntimeError("Previous response ID cannot be used in conjunction with conversation ID.") - context_id = request.previous_response_id or context.conversation_id - - # The following should never happen due to the checks above. - # This is for type safety and defensive programming. - if self._checkpoint_storage_path is None: - raise RuntimeError("Checkpoint storage path is not configured for workflow agent.") - if not isinstance(self._agent, WorkflowAgent): - raise RuntimeError("Agent is not a workflow agent.") - - # Workflow agents are not async context managers in any built-in path, - # but call _ensure_agent_ready for symmetry with the regular path so - # any future async resources owned by the workflow are entered here. - await self._ensure_agent_ready() - - # Determine the latest checkpoint (if any) so we can resume the - # workflow's prior state for this turn. The directory is keyed by - # the inbound context id (conversation_id when set, otherwise - # previous_response_id). Multi-turn declarative workflows need the - # workflow's internal state (e.g. Conversation.messages, - # intermediate Local.* variables) to survive across user turns; - # the only place that state lives is the workflow checkpoint, so - # on every turn we restore the latest checkpoint and feed the new - # input back into the start executor as a continuation rather than - # a fresh run. - latest_checkpoint_id: str | None = None - restore_storage: FileCheckpointStorage | None = None - if context_id is not None: - restore_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, context_id) - latest_checkpoint = await restore_storage.get_latest(workflow_name=self._agent.workflow.name) - if latest_checkpoint is not None: - latest_checkpoint_id = latest_checkpoint.checkpoint_id - - # Storage that will receive checkpoints written during this turn. - # When the caller chains with previous_response_id, the next turn - # will reference the current response_id as its previous_response_id, - # so new checkpoints must land under the current response_id (or the - # conversation_id when set). When conversation_id is set, this - # matches restore_storage; when only previous_response_id was - # supplied, restore_storage points at the *prior* response's - # directory and checkpoint_storage points at the *current* response's. - write_context_id = context.conversation_id or context.response_id - checkpoint_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, write_context_id) - - # Multi-turn pattern: when we have a prior checkpoint, restore it - # first (drive the workflow back to idle with prior state intact), - # then make a separate call that delivers the new user input. This - # depends on Workflow.run preserving shared state across calls. The - # restore-only call may yield events from any pending in-flight - # work in the checkpoint; we consume those internally here so they - # don't surface to the response stream as duplicates. - # - # If the restored checkpoint had pending request_info events, the - # restore-only call replays them through - # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates`` - # and populates ``self._agent.pending_requests``. That is the correct - # state: those requests are genuinely outstanding, and the next - # ``run(input_messages, ...)`` call may contain ``function_call_output`` - # items (carried as FunctionResult/FunctionApprovalResponse content) - # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. - if latest_checkpoint_id is not None: - if restore_storage is None: # pragma: no cover - defensive - raise RuntimeError("Checkpoint restore storage is not configured.") - if is_streaming_request: - async for _ in self._agent.run( - stream=True, - checkpoint_id=latest_checkpoint_id, - checkpoint_storage=restore_storage, - ): - pass - else: - await self._agent.run( - stream=False, - checkpoint_id=latest_checkpoint_id, - checkpoint_storage=restore_storage, - ) - - # Now run the agent with the latest input + """Handle the creation of a response for a workflow agent.""" response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model) - yield response_event_stream.emit_created() yield response_event_stream.emit_in_progress() - if not is_streaming_request: - # Run the agent in non-streaming mode - response = await self._agent.run(input_messages, stream=False, checkpoint_storage=checkpoint_storage) - - for message in response.messages: - for content in message.contents: - async for item in _to_outputs( - response_event_stream, - content, - approval_storage=self._approval_storage, - ): - yield item - - await self._delete_not_latest_checkpoints(checkpoint_storage, self._agent.workflow.name) - yield response_event_stream.emit_completed() - return - # Track the current active output item builder for streaming; # lazily created on matching content, closed when a different type arrives. - tracker = _OutputItemTracker(response_event_stream) + tracker: _OutputItemTracker | None = None - # Run the workflow agent in streaming mode - async for update in self._agent.run(input_messages, stream=True, checkpoint_storage=checkpoint_storage): - for content in update.contents: - for event in tracker.handle(content): - yield event - if tracker.needs_async: - async for item in _to_outputs( - response_event_stream, - content, - approval_storage=self._approval_storage, + try: + input_items = await context.get_input_items() + input_messages = await _items_to_messages(input_items, approval_storage=self._approval_storage) + is_streaming_request = request.stream is not None and request.stream is True + + _, are_options_set = _to_chat_options(request) + if are_options_set: + logger.warning("Workflow agent doesn't support runtime options. They will be ignored.") + + if request.previous_response_id is not None and context.conversation_id is not None: + raise RuntimeError("Previous response ID cannot be used in conjunction with conversation ID.") + context_id = request.previous_response_id or context.conversation_id + + # The following should never happen due to the checks above. + # This is for type safety and defensive programming. + if self._checkpoint_storage_path is None: + raise RuntimeError("Checkpoint storage path is not configured for workflow agent.") + if not isinstance(self._agent, WorkflowAgent): + raise RuntimeError("Agent is not a workflow agent.") + + # Workflow agents are not async context managers in any built-in path, + # but call _ensure_agent_ready for symmetry with the regular path so + # any future async resources owned by the workflow are entered here. + await self._ensure_agent_ready() + + # Determine the latest checkpoint (if any) so we can resume the + # workflow's prior state for this turn. The directory is keyed by + # the inbound context id (conversation_id when set, otherwise + # previous_response_id). Multi-turn declarative workflows need the + # workflow's internal state (e.g. Conversation.messages, + # intermediate Local.* variables) to survive across user turns; + # the only place that state lives is the workflow checkpoint, so + # on every turn we restore the latest checkpoint and feed the new + # input back into the start executor as a continuation rather than + # a fresh run. + latest_checkpoint_id: str | None = None + restore_storage: FileCheckpointStorage | None = None + if context_id is not None: + restore_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, context_id) + latest_checkpoint = await restore_storage.get_latest(workflow_name=self._agent.workflow.name) + if latest_checkpoint is not None: + latest_checkpoint_id = latest_checkpoint.checkpoint_id + + # Storage that will receive checkpoints written during this turn. + # When the caller chains with previous_response_id, the next turn + # will reference the current response_id as its previous_response_id, + # so new checkpoints must land under the current response_id (or the + # conversation_id when set). When conversation_id is set, this + # matches restore_storage; when only previous_response_id was + # supplied, restore_storage points at the *prior* response's + # directory and write_storage points at the *current* response's. + write_context_id = context.conversation_id or context.response_id + write_storage = _checkpoint_storage_for_context(self._checkpoint_storage_path, write_context_id) + + # Multi-turn pattern: when we have a prior checkpoint, restore it + # first (drive the workflow back to idle with prior state intact), + # then make a separate call that delivers the new user input. This + # depends on Workflow.run preserving shared state across calls. The + # restore-only call may yield events from any pending in-flight + # work in the checkpoint; we consume those internally here so they + # don't surface to the response stream as duplicates. + # + # If the restored checkpoint had pending request_info events, the + # restore-only call replays them through + # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates`` + # and populates ``self._agent.pending_requests``. That is the correct + # state: those requests are genuinely outstanding, and the next + # ``run(input_messages, ...)`` call may contain ``function_call_output`` + # items (carried as FunctionResult/FunctionApprovalResponse content) + # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. + if latest_checkpoint_id is not None: + if is_streaming_request: + async for _ in self._agent.run( + stream=True, + checkpoint_id=latest_checkpoint_id, + checkpoint_storage=restore_storage, ): - yield item - tracker.needs_async = False + pass + else: + await self._agent.run( + stream=False, + checkpoint_id=latest_checkpoint_id, + checkpoint_storage=restore_storage, + ) - # Close any remaining active builder - for event in tracker.close(): - yield event + if not is_streaming_request: + # Run the agent in non-streaming mode with the new user input. + response = await self._agent.run( + input_messages, + stream=False, + checkpoint_storage=write_storage, + ) - await self._delete_not_latest_checkpoints(checkpoint_storage, self._agent.workflow.name) - yield response_event_stream.emit_completed() + async for item in _to_outputs_for_messages( + response_event_stream, + response.messages, + approval_storage=self._approval_storage, + ): + yield item + + await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name) + yield response_event_stream.emit_completed() + return + + tracker = _OutputItemTracker(response_event_stream) + + # Run the workflow agent in streaming mode with the new user input. + async for update in self._agent.run( + input_messages, + stream=True, + checkpoint_storage=write_storage, + ): + for content in update.contents: + for event in tracker.handle(content): + yield event + if tracker.needs_async: + async for item in _to_outputs( + response_event_stream, content, approval_storage=self._approval_storage + ): + yield item + tracker.needs_async = False + + # Close any remaining active builder + for event in tracker.close(): + yield event + + await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name) + yield response_event_stream.emit_completed() + except Exception as ex: + logger.exception("Failed to produce response for workflow agent") + for event in self._emit_failure(response_event_stream, tracker, ex): + yield event @staticmethod async def _delete_not_latest_checkpoints(checkpoint_storage: FileCheckpointStorage, workflow_name: str) -> None: @@ -614,6 +711,29 @@ class ResponsesHostServer(ResponsesAgentServerHost): if checkpoint.checkpoint_id != latest_checkpoint.checkpoint_id: await checkpoint_storage.delete(checkpoint.checkpoint_id) + @staticmethod + def _emit_failure( + response_event_stream: ResponseEventStream, + tracker: _OutputItemTracker | None, + ex: BaseException, + ) -> Generator[ResponseStreamEvent]: + """Yield a terminal ``response.failed`` event for ``ex``. + + Drains any in-progress streaming output item first so the resulting + SSE stream stays well-formed, then emits ``response.failed`` carrying + the exception's message (falling back to the exception type name when + ``str(ex)`` is empty). Any error raised while draining the tracker is + logged and otherwise ignored so that the original failure is always + what the client sees. + """ + if tracker is not None: + try: + yield from tracker.close() + except Exception: + logger.exception("Error while closing streaming tracker after failure") + message = str(ex) or type(ex).__name__ + yield response_event_stream.emit_failed(message=message) + # endregion ResponsesHostServer @@ -676,7 +796,7 @@ class _OutputItemTracker: 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}" + key = content.call_id or 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) @@ -685,6 +805,24 @@ class _OutputItemTracker: if self._mcp_builder is not None: yield self._mcp_builder.emit_arguments_delta(args_str) + elif ( + content.type == "mcp_server_tool_result" + and self._active_type == "mcp_server_tool_call" + and self._mcp_builder is not None + and content.call_id is not None + and content.call_id == self._mcp_builder.item_id + ): + accumulated = "".join(self._accumulated) + yield self._mcp_builder.emit_arguments_done(accumulated) + yield self._mcp_builder.emit_completed() + yield self._mcp_builder.emit_done(output=_stringify_mcp_output(content.output)) + self._mcp_builder = None + self._active_type = None + self._active_id = None + self._accumulated.clear() + self.needs_async = False + return + else: yield from self._close() self.needs_async = True @@ -724,9 +862,10 @@ class _OutputItemTracker: self._mcp_builder = self._stream.add_output_item_mcp_call( server_label=content.server_name or "default", name=content.tool_name or "", + item_id=content.call_id, ) self._active_type = "mcp_server_tool_call" - self._active_id = f"{content.server_name or 'default'}::{content.tool_name}" + self._active_id = content.call_id or f"{content.server_name or 'default'}::{content.tool_name}" yield self._mcp_builder.emit_added() def _close(self) -> Generator[ResponseStreamEvent]: @@ -762,9 +901,6 @@ class _OutputItemTracker: self._accumulated.clear() -# endregion - - # region Option Conversion @@ -800,6 +936,695 @@ def _to_chat_options(request: CreateResponse) -> tuple[ChatOptions, bool]: # endregion +# region Input Message Conversion + + +async def _items_to_messages( + input_items: Sequence[Item], *, approval_storage: ApprovalStorage | None = None +) -> list[Message]: + """Converts a sequence of input items to a list of Messages, one per item. + + Args: + input_items: The input items to convert. + approval_storage: An optional ApprovalStorage instance used to look up + approval requests when converting MCP approval response items. + + Returns: + A list of Messages, one per supported input item. + """ + messages: list[Message] = [] + for item in input_items: + messages.append(await _item_to_message(item, approval_storage=approval_storage)) + return messages + + +async def _item_to_message(item: Item, *, approval_storage: ApprovalStorage | None = None) -> Message: + """Converts an Item to a Message. + + Args: + item: The Item to convert. + approval_storage: An optional ApprovalStorage instance used to look up + approval requests when converting MCP approval response items. + + Returns: + The converted Message. + + Raises: + ValueError: If the Item type is not supported. + """ + if item.type == "message": + msg = cast(ItemMessage, item) + if isinstance(msg.content, str): + return Message(role=msg.role, contents=[Content.from_text(msg.content)]) + return Message(role=msg.role, contents=[_convert_message_content(part) for part in msg.content]) + + if item.type == "output_message": + output_msg = cast(ItemOutputMessage, item) + return Message( + role=output_msg.role, contents=[_convert_output_message_content(part) for part in output_msg.content] + ) + + if item.type == "function_call": + fc = cast(ItemFunctionToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(fc.call_id, fc.name, arguments=fc.arguments)], + ) + + if item.type == "function_call_output": + fco = cast(FunctionCallOutputItemParam, item) + output = fco.output if isinstance(fco.output, str) else str(fco.output) + return Message( + role="tool", + contents=[Content.from_function_result(fco.call_id, result=output)], + ) + + if item.type == "reasoning": + reasoning = cast(ItemReasoningItem, item) + reason_contents: list[Content] = [] + if reasoning.summary: + for summary in reasoning.summary: + reason_contents.append(Content.from_text(summary.text)) + return Message(role="assistant", contents=reason_contents) + + if item.type == "mcp_call": + mcp = cast(ItemMcpToolCall, item) + contents = [ + Content.from_mcp_server_tool_call( + mcp.id, + mcp.name, + server_name=mcp.server_label, + arguments=mcp.arguments, + ) + ] + if getattr(mcp, "output", None) is not None: + contents.append(Content.from_mcp_server_tool_result(call_id=mcp.id, output=mcp.output)) + return Message( + role="assistant", + contents=contents, + ) + + if item.type == "mcp_approval_request": + mcp_req = cast(ItemMcpApprovalRequest, item) + if approval_storage is not None: + function_approval_request_content = await approval_storage.load_approval_request(mcp_req.id) + else: + raise ValueError("ApprovalStorage is required to load approval request.") + return Message( + role="assistant", + contents=[function_approval_request_content], + ) + + if item.type == "mcp_approval_response": + mcp_resp = cast(MCPApprovalResponse, item) + if approval_storage is not None: + function_approval_request_content = await approval_storage.load_approval_request( + mcp_resp.approval_request_id + ) + else: + raise ValueError("ApprovalStorage is required to load approval request.") + return Message( + role="user", + contents=[function_approval_request_content.to_function_approval_response(mcp_resp.approve)], + ) + + if item.type == "code_interpreter_call": + ci = cast(ItemCodeInterpreterToolCall, item) + return Message( + role="assistant", + contents=[Content.from_code_interpreter_tool_call(call_id=ci.id)], + ) + + if item.type == "image_generation_call": + ig = cast(ItemImageGenToolCall, item) + return Message( + role="assistant", + contents=[Content.from_image_generation_tool_call(image_id=ig.id)], + ) + + if item.type == "shell_call": + sc = cast(FunctionShellCallItemParam, item) + return Message( + role="assistant", + contents=[ + Content.from_shell_tool_call( + call_id=sc.call_id, + commands=sc.action.commands, + status=str(sc.status), + ) + ], + ) + + if item.type == "shell_call_output": + sco = cast(FunctionShellCallOutputItemParam, item) + outputs = [ + Content.from_shell_command_output( + stdout=out.stdout or "", + stderr=out.stderr or "", + exit_code=getattr(out.outcome, "exit_code", None) if hasattr(out, "outcome") else None, + ) + for out in (sco.output or []) + ] + return Message( + role="tool", + contents=[ + Content.from_shell_tool_result( + call_id=sco.call_id, + outputs=outputs, + max_output_length=sco.max_output_length, + ) + ], + ) + + if item.type == "local_shell_call": + lsc = cast(ItemLocalShellToolCall, item) + commands = lsc.action.command if hasattr(lsc.action, "command") and lsc.action.command else [] + return Message( + role="assistant", + contents=[ + Content.from_shell_tool_call( + call_id=lsc.call_id, + commands=commands, + status=str(lsc.status), + ) + ], + ) + + if item.type == "local_shell_call_output": + lsco = cast(ItemLocalShellToolCallOutput, item) + return Message( + role="tool", + contents=[ + Content.from_shell_tool_result( + call_id=lsco.id, + outputs=[Content.from_shell_command_output(stdout=lsco.output)], + ) + ], + ) + + if item.type == "file_search_call": + fs = cast(ItemFileSearchToolCall, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + fs.id, + "file_search", + arguments=json.dumps({"queries": fs.queries}), + ) + ], + ) + + if item.type == "web_search_call": + ws = cast(ItemWebSearchToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(ws.id, "web_search")], + ) + + if item.type == "computer_call": + cc = cast(ItemComputerToolCall, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + cc.call_id, + "computer_use", + arguments=str(cc.action), + ) + ], + ) + + if item.type == "computer_call_output": + cco = cast(ComputerCallOutputItemParam, item) + return Message( + role="tool", + contents=[Content.from_function_result(cco.call_id, result=str(cco.output))], + ) + + if item.type == "custom_tool_call": + ct = cast(ItemCustomToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(ct.call_id, ct.name, arguments=ct.input)], + ) + + if item.type == "custom_tool_call_output": + cto = cast(ItemCustomToolCallOutput, item) + output = cto.output if isinstance(cto.output, str) else str(cto.output) + # Hosted-MCP results land here because the host writes them via + # `aoutput_item_custom_tool_call_output` (see `_to_outputs` for + # `mcp_server_tool_result`). The persisted `call_id` keeps its + # `mcp_*` prefix; on read, route those back to a hosted-MCP result + # Content so the chat-client serialize layer can coalesce them + # onto a single `mcp_call` input item with `output` populated. + # Issue #5546. + if cto.call_id and cto.call_id.startswith("mcp_"): + return Message( + role="tool", + contents=[Content.from_mcp_server_tool_result(call_id=cto.call_id, output=output)], + ) + return Message( + role="tool", + contents=[Content.from_function_result(cto.call_id, result=output)], + ) + + if item.type == "apply_patch_call": + ap = cast(ApplyPatchToolCallItemParam, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + ap.call_id, + "apply_patch", + arguments=str(ap.operation), + ) + ], + ) + + if item.type == "apply_patch_call_output": + apo = cast(ApplyPatchToolCallOutputItemParam, item) + return Message( + role="tool", + contents=[Content.from_function_result(apo.call_id, result=apo.output or "")], + ) + + raise ValueError(f"Unsupported Item type: {item.type}") + + +async def _output_items_to_messages( + history: Sequence[OutputItem], + *, + approval_storage: ApprovalStorage | None = None, +) -> list[Message]: + """Converts a sequence of OutputItem objects to a list of Message objects. + + Args: + history (Sequence[OutputItem]): The sequence of OutputItem objects to convert. + approval_storage (ApprovalStorage | None, optional): The approval storage to use for + resolving MCP approval requests. Defaults to None. + + Returns: + list[Message]: The list of Message objects. + """ + messages: list[Message] = [] + for item in history: + messages.append(await _output_item_to_message(item, approval_storage=approval_storage)) + return messages + + +async def _output_item_to_message(item: OutputItem, *, approval_storage: ApprovalStorage | None = None) -> Message: + """Converts an OutputItem to a Message. + + Args: + item (OutputItem): The OutputItem to convert. + approval_storage (ApprovalStorage | None, optional): The approval storage to use for + resolving MCP approval requests. Defaults to None. + + Returns: + Message: The converted Message. + + Raises: + ValueError: If the OutputItem type is not supported. + """ + if item.type == "output_message": + output_msg = cast(OutputItemOutputMessage, item) + return Message( + role=output_msg.role, contents=[_convert_output_message_content(part) for part in output_msg.content] + ) + + if item.type == "message": + msg = cast(OutputItemMessage, item) + return Message(role=msg.role, contents=[_convert_message_content(part) for part in msg.content]) + + if item.type == "function_call": + fc = cast(OutputItemFunctionToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(fc.call_id, fc.name, arguments=fc.arguments)], + ) + + if item.type == "function_call_output": + fco = cast(FunctionCallOutputItemParam, item) + output = fco.output if isinstance(fco.output, str) else str(fco.output) + return Message( + role="tool", + contents=[Content.from_function_result(fco.call_id, result=output)], + ) + + if item.type == "reasoning": + reasoning = cast(OutputItemReasoningItem, item) + contents: list[Content] = [] + if reasoning.summary: + for summary in reasoning.summary: + contents.append(Content.from_text(summary.text)) + return Message(role="assistant", contents=contents) + + if item.type == "mcp_call": + mcp = cast(OutputItemMcpToolCall, item) + contents = [ + Content.from_mcp_server_tool_call( + mcp.id, + mcp.name, + server_name=mcp.server_label, + arguments=mcp.arguments, + ) + ] + if getattr(mcp, "output", None) is not None: + contents.append(Content.from_mcp_server_tool_result(call_id=mcp.id, output=mcp.output)) + return Message( + role="assistant", + contents=contents, + ) + + if item.type == "mcp_approval_request": + mcp_req = cast(OutputItemMcpApprovalRequest, item) + if approval_storage is not None: + function_approval_request_content = await approval_storage.load_approval_request(mcp_req.id) + else: + raise ValueError("ApprovalStorage is required to load approval request.") + return Message( + role="assistant", + contents=[function_approval_request_content], + ) + + if item.type == "mcp_approval_response": + mcp_resp = cast(OutputItemMcpApprovalResponseResource, item) + if approval_storage is not None: + function_approval_request_content = await approval_storage.load_approval_request( + mcp_resp.approval_request_id + ) + else: + raise ValueError("ApprovalStorage is required to load approval request.") + + return Message( + role="user", + contents=[function_approval_request_content.to_function_approval_response(mcp_resp.approve)], + ) + + if item.type == "code_interpreter_call": + ci = cast(OutputItemCodeInterpreterToolCall, item) + return Message( + role="assistant", + contents=[Content.from_code_interpreter_tool_call(call_id=ci.id)], + ) + + if item.type == "image_generation_call": + ig = cast(OutputItemImageGenToolCall, item) + return Message( + role="assistant", + contents=[Content.from_image_generation_tool_call(image_id=ig.id)], + ) + + if item.type == "shell_call": + sc = cast(OutputItemFunctionShellCall, item) + return Message( + role="assistant", + contents=[ + Content.from_shell_tool_call( + call_id=sc.call_id, + commands=sc.action.commands, + status=str(sc.status), + ) + ], + ) + + if item.type == "shell_call_output": + sco = cast(OutputItemFunctionShellCallOutput, item) + outputs = [ + Content.from_shell_command_output( + stdout=out.stdout or "", + stderr=out.stderr or "", + exit_code=getattr(out.outcome, "exit_code", None) if hasattr(out, "outcome") else None, + ) + for out in (sco.output or []) + ] + return Message( + role="tool", + contents=[ + Content.from_shell_tool_result( + call_id=sco.call_id, + outputs=outputs, + max_output_length=sco.max_output_length, + ) + ], + ) + + if item.type == "local_shell_call": + lsc = cast(OutputItemLocalShellToolCall, item) + commands = lsc.action.command if hasattr(lsc.action, "command") and lsc.action.command else [] + return Message( + role="assistant", + contents=[ + Content.from_shell_tool_call( + call_id=lsc.call_id, + commands=commands, + status=str(lsc.status), + ) + ], + ) + + if item.type == "local_shell_call_output": + lsco = cast(OutputItemLocalShellToolCallOutput, item) + return Message( + role="tool", + contents=[ + Content.from_shell_tool_result( + call_id=lsco.id, + outputs=[Content.from_shell_command_output(stdout=lsco.output)], + ) + ], + ) + + if item.type == "file_search_call": + fs = cast(OutputItemFileSearchToolCall, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + fs.id, + "file_search", + arguments=json.dumps({"queries": fs.queries}), + ) + ], + ) + + if item.type == "web_search_call": + ws = cast(OutputItemWebSearchToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(ws.id, "web_search")], + ) + + if item.type == "computer_call": + cc = cast(OutputItemComputerToolCall, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + cc.call_id, + "computer_use", + arguments=str(cc.action), + ) + ], + ) + + if item.type == "computer_call_output": + cco = cast(OutputItemComputerToolCallOutputResource, item) + return Message( + role="tool", + contents=[Content.from_function_result(cco.call_id, result=str(cco.output))], + ) + + if item.type == "custom_tool_call": + ct = cast(OutputItemCustomToolCall, item) + return Message( + role="assistant", + contents=[Content.from_function_call(ct.call_id, ct.name, arguments=ct.input)], + ) + + if item.type == "custom_tool_call_output": + cto = cast(OutputItemCustomToolCallOutput, item) + output = cto.output if isinstance(cto.output, str) else str(cto.output) + # Hosted-MCP results land here because the host writes them via + # `aoutput_item_custom_tool_call_output`. Route `mcp_*` call_ids + # back to a hosted-MCP result Content so the chat-client serialize + # layer can coalesce onto the matching `mcp_call` input item. + # Issue #5546. + if cto.call_id and cto.call_id.startswith("mcp_"): + return Message( + role="tool", + contents=[Content.from_mcp_server_tool_result(call_id=cto.call_id, output=output)], + ) + return Message( + role="tool", + contents=[Content.from_function_result(cto.call_id, result=output)], + ) + + if item.type == "apply_patch_call": + ap = cast(OutputItemApplyPatchToolCall, item) + return Message( + role="assistant", + contents=[ + Content.from_function_call( + ap.call_id, + "apply_patch", + arguments=str(ap.operation), + ) + ], + ) + + if item.type == "apply_patch_call_output": + apo = cast(OutputItemApplyPatchToolCallOutput, item) + return Message( + role="tool", + contents=[Content.from_function_result(apo.call_id, result=apo.output or "")], + ) + + if item.type == "oauth_consent_request": + oauth = cast(OAuthConsentRequestOutputItem, item) + return Message( + role="assistant", + contents=[Content.from_oauth_consent_request(oauth.consent_link)], + ) + + if item.type == "structured_outputs": + so = cast(StructuredOutputsOutputItem, item) + text = json.dumps(so.output) if not isinstance(so.output, str) else so.output + return Message(role="assistant", contents=[Content.from_text(text)]) + + raise ValueError(f"Unsupported OutputItem type: {item.type}") + + +def _convert_output_message_content(content: OutputMessageContent) -> Content: + """Converts an OutputMessageContent to a Content object. + + Args: + content (OutputMessageContent): The OutputMessageContent to convert. + + Returns: + Content: The converted Content object. + + Raises: + ValueError: If the OutputMessageContent type is not supported. + """ + if content.type == "output_text": + text_content = cast(OutputMessageContentOutputTextContent, content) + return Content.from_text(text_content.text) + if content.type == "refusal": + refusal_content = cast(OutputMessageContentRefusalContent, content) + return Content.from_text(refusal_content.refusal) + + raise ValueError(f"Unsupported OutputMessageContent type: {content.type}") + + +def _convert_file_data(data_uri: str, filename: str | None = None) -> Content: + """Convert a file_data data URI to a Content object. + + For text/* MIME types, decodes the base64 content and returns it as text. + For other types, returns a URI-based Content with the filename preserved. + """ + # Parse data URI: data:;base64, + if data_uri.startswith("data:") and ";base64," in data_uri: + header, encoded = data_uri.split(";base64,", 1) + media_type = header[len("data:") :] + if media_type.startswith("text/"): + try: + decoded_text = base64.b64decode(encoded).decode("utf-8") + except (ValueError, UnicodeDecodeError): + logger.warning( + "Failed to decode text/* file_data as UTF-8, falling through to URI passthrough.", + exc_info=True, + ) + else: + prefix = f"[File: {filename}]\n" if filename else "" + return Content.from_text(f"{prefix}{decoded_text}") + additional_properties = {"filename": filename} if filename else None + return Content.from_uri(data_uri, additional_properties=additional_properties) + + +def _convert_message_content(content: MessageContent) -> Content: + """Converts a MessageContent to a Content object. + + Args: + content (MessageContent): The MessageContent to convert. + + Returns: + Content: The converted Content object. + + Raises: + ValueError: If the MessageContent type is not supported. + """ + if content.type == "input_text": + input_text = cast(MessageContentInputTextContent, content) + return Content.from_text(input_text.text) + if content.type == "output_text": + output_text = cast(MessageContentOutputTextContent, content) + return Content.from_text(output_text.text) + if content.type == "text": + text = cast(TextContent, content) + return Content.from_text(text.text) + if content.type == "summary_text": + summary = cast(SummaryTextContent, content) + return Content.from_text(summary.text) + if content.type == "refusal": + refusal = cast(MessageContentRefusalContent, content) + return Content.from_text(refusal.refusal) + if content.type == "reasoning_text": + reasoning = cast(MessageContentReasoningTextContent, content) + return Content.from_text_reasoning(text=reasoning.text) + if content.type == "input_image": + image = cast(MessageContentInputImageContent, content) + if image.image_url: + if image.image_url.startswith("data:"): + return Content.from_uri(image.image_url) + return Content.from_uri(image.image_url, media_type="image/*") + if image.file_id: + return Content.from_hosted_file(image.file_id) + if content.type == "input_file": + file = cast(MessageContentInputFileContent, content) + if file.file_url: + return Content.from_uri(file.file_url) + if file.file_id: + return Content.from_hosted_file(file.file_id, name=file.filename) + if file.file_data: + return _convert_file_data(file.file_data, file.filename) + if content.type == "computer_screenshot": + screenshot = cast(ComputerScreenshotContent, content) + return Content.from_uri(screenshot.image_url) + + raise ValueError(f"Unsupported MessageContent type: {content.type}") + + +# endregion + +# region Output Item Conversion + + +def _argument_json_default(value: Any) -> Any: + if is_dataclass(value) and not isinstance(value, type): + return asdict(value) + to_dict = getattr(value, "to_dict", None) + if callable(to_dict): + return to_dict() + raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") + + +def _arguments_to_str(arguments: Any | None) -> str: + """Convert arguments to a JSON string. + + Args: + arguments: The arguments to convert, can be a string, JSON-like object, or None. + + Returns: + The arguments as a JSON string. + """ + if arguments is None: + return "" + if isinstance(arguments, str): + return arguments + return json.dumps(arguments, default=_argument_json_default) + async def _to_outputs( stream: ResponseEventStream, @@ -846,6 +1671,7 @@ async def _to_outputs( mcp_call = stream.add_output_item_mcp_call( server_label=content.server_name or "default", name=content.tool_name or "", + item_id=content.call_id, ) yield mcp_call.emit_added() async for event in mcp_call.aarguments(_arguments_to_str(content.arguments)): @@ -920,4 +1746,91 @@ async def _to_outputs( logger.warning(f"Content type '{content.type}' is not supported yet. This is usually safe to ignore.") +def _stringify_mcp_output(output: Any) -> str: + """Convert hosted MCP output payloads into the string shape expected by mcp_call.output.""" + if output is None: + return "" + if isinstance(output, str): + return output + if isinstance(output, Mapping): + text = cast(Any, output).get("text") + if isinstance(text, str): + return text + return json.dumps(output, default=str) + if isinstance(output, Sequence) and not isinstance(output, (str, bytes, bytearray)): + parts: list[str] = [] + entries = cast(Sequence[object], output) + for entry in entries: + if isinstance(entry, Content) and entry.type == "text": + parts.append(entry.text or "") + continue + parts.append(_stringify_mcp_output(entry)) + return "".join(parts) + return str(output) + + +def _emit_completed_mcp_call( + stream: ResponseEventStream, + call_content: Content, + *, + arguments: str, + output: str, +) -> Generator[ResponseStreamEvent]: + """Emit a single completed MCP call item carrying both arguments and output.""" + mcp_call = stream.add_output_item_mcp_call( + server_label=call_content.server_name or "default", + name=call_content.tool_name or "", + item_id=call_content.call_id, + ) + yield mcp_call.emit_added() + yield mcp_call.emit_arguments_done(arguments) + yield mcp_call.emit_completed() + yield mcp_call.emit_done(output=output) + + +async def _to_outputs_for_messages( + stream: ResponseEventStream, + messages: Sequence[Message], + *, + approval_storage: ApprovalStorage | None = None, +) -> AsyncIterator[ResponseStreamEvent]: + """Convert messages to output events with hosted-MCP call/result coalescing. + + Parse once in message/content order and emit either: + - a single canonical completed ``mcp_call`` when adjacent hosted MCP + call/result content are encountered, or + - standard output items for all other content types. + """ + pending_mcp_call: Content | None = None + + for message in messages: + for content in message.contents: + if pending_mcp_call is not None: + if content.type == "mcp_server_tool_result" and content.call_id == pending_mcp_call.call_id: + for event in _emit_completed_mcp_call( + stream, + pending_mcp_call, + arguments=_arguments_to_str(pending_mcp_call.arguments), + output=_stringify_mcp_output(content.output), + ): + yield event + pending_mcp_call = None + continue + + async for event in _to_outputs(stream, pending_mcp_call, approval_storage=approval_storage): + yield event + pending_mcp_call = None + + if content.type == "mcp_server_tool_call" and content.call_id: + pending_mcp_call = content + continue + + async for event in _to_outputs(stream, content, approval_storage=approval_storage): + yield event + + if pending_mcp_call is not None: + async for event in _to_outputs(stream, pending_mcp_call, approval_storage=approval_storage): + yield event + + # endregion diff --git a/python/packages/foundry_hosting/tests/test_history_provider.py b/python/packages/foundry_hosting/tests/test_history_provider.py index cfdbeccac..6ee8a63cb 100644 --- a/python/packages/foundry_hosting/tests/test_history_provider.py +++ b/python/packages/foundry_hosting/tests/test_history_provider.py @@ -913,8 +913,9 @@ class TestSharedReExports: downstream code that historically imported them keep working.""" def test_responses_re_exports_helpers(self) -> None: - # All of these used to live in ``_responses``; after the - # refactor they live in ``_shared`` but are re-exported. + # These helpers historically lived in ``_responses``. They must + # remain importable there for compatibility even when ``_shared`` + # also provides canonical implementations for the history provider. from agent_framework_foundry_hosting import ( _responses, # pyright: ignore[reportPrivateUsage] _shared, # pyright: ignore[reportPrivateUsage] @@ -929,9 +930,8 @@ class TestSharedReExports: "_output_item_to_message", "_output_items_to_messages", ): - assert getattr(_responses, name) is getattr(_shared, name), ( - f"{name} should be re-exported from _responses for backwards compat" - ) + assert callable(getattr(_responses, name)) + assert callable(getattr(_shared, name)) # region Full AF ↔ Foundry round-trip via InMemoryResponseProvider diff --git a/python/packages/hosting/agent_framework_hosting/_host.py b/python/packages/hosting/agent_framework_hosting/_host.py index c0a6a8e46..3d5287f38 100644 --- a/python/packages/hosting/agent_framework_hosting/_host.py +++ b/python/packages/hosting/agent_framework_hosting/_host.py @@ -224,7 +224,7 @@ def _workflow_event_to_update(event: WorkflowEvent[Any]) -> AgentResponseUpdate @asynccontextmanager -async def _suppress_already_consumed() -> AsyncIterator[None]: # noqa: RUF029 +async def _suppress_already_consumed() -> AsyncIterator[None]: """Yield, swallowing finalizer failures so consumer cleanup never crashes the host. The bridge stream calls ``get_final_response()`` after iterating the