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 888eee297..72fcfbacd 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -376,16 +376,24 @@ class ResponsesHostServer(ResponsesAgentServerHost): # is created for this request and stored under the current response_id. The current response_id # will become the previous_response_id for the next request in a response chain, allowing the # session to be retrieved. - session_id = context.conversation_id or request.get("previous_response_id") - session = await session_storage.get(session_id) if session_id is not None else None - if session is None: - if session_id is not None: + if (previous_response_id := request.get("previous_response_id")) is not None: + session = await session_storage.get(previous_response_id) + if session is None: + raise RuntimeError( + f"Cannot find an existing agent session for previous_response_id={previous_response_id}. " + "Ensure that the previous response was created successfully and that the ID is correct." + ) + elif (conversation_id := context.conversation_id) is not None: + session = await session_storage.get(conversation_id) + if session is None: # Note that we cannot determine if the session was deleted or never existed, # so we log a warning and create a new session. logger.info( "Cannot find an existing agent session for id=%s. Creating a new session.", - session_id, + conversation_id, ) + session = self._agent.create_session() + else: session = self._agent.create_session() except Exception as ex: logger.error("Failed to prepare state storage: %s", ex, exc_info=(type(ex), ex, ex.__traceback__)) diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index 634dc26f7..a7d638210 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -157,9 +157,8 @@ def _make_agent( def run_streaming(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: del args - if kwargs.get("stream"): - return ResponseStream(_stream_gen(), finalizer=AgentResponse.from_updates) - raise NotImplementedError("Only streaming is configured on this mock") + assert kwargs.get("stream") is True + return ResponseStream(_stream_gen(), finalizer=AgentResponse.from_updates) agent.run = MagicMock(side_effect=run_streaming) @@ -180,19 +179,13 @@ class _RecordingHistoryClient(BaseChatClient): **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: del options, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." self.calls.append(list(messages)) - if stream: + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=[Content.from_text("recorded")], role="assistant") - async def stream_response() -> AsyncIterator[ChatResponseUpdate]: - yield ChatResponseUpdate(contents=[Content.from_text("recorded")], role="assistant") - - return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) - - async def get_response() -> ChatResponse: - return ChatResponse(messages=[Message(role="assistant", contents=[Content.from_text("recorded")])]) - - return get_response() + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) class _PerServiceCallHistoryProvider(HistoryProvider): @@ -238,30 +231,27 @@ class _FunctionLoopRecordingClient( **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: del options, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." self.calls.append(list(messages)) self.saves_before_call.append(self._provider.save_calls) call_number = len(self.calls) - if stream: + async def stream_response() -> AsyncIterator[ChatResponseUpdate]: + if call_number == 1: + yield ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="call_1", + name="lookup_weather", + arguments='{"location": "Seattle"}', + ) + ], + role="assistant", + ) + else: + yield ChatResponseUpdate(contents=[Content.from_text("It is sunny in Seattle.")], role="assistant") - async def stream_response() -> AsyncIterator[ChatResponseUpdate]: - if call_number == 1: - yield ChatResponseUpdate( - contents=[ - Content.from_function_call( - call_id="call_1", - name="lookup_weather", - arguments='{"location": "Seattle"}', - ) - ], - role="assistant", - ) - else: - yield ChatResponseUpdate(contents=[Content.from_text("It is sunny in Seattle.")], role="assistant") - - return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) - - raise NotImplementedError("The inner agent only runs in stream mode in Foundry Hosted Agents.") + return ResponseStream(stream_response(), finalizer=ChatResponse.from_updates) @tool(name="lookup_weather", approval_mode="never_require") @@ -503,7 +493,7 @@ class TestResponsesHostServerInit: failed_event = cast(Mapping[str, Any], failed_events[0]) response = cast(Mapping[str, Any], failed_event["response"]) error = cast(Mapping[str, Any], response["error"]) - assert "No Agent Framework session was found" in error["message"] + assert "Cannot find an existing agent session for previous_response_id=response-missing." in error["message"] agent.run.assert_not_called() agent.create_session.assert_not_called() @@ -3901,9 +3891,8 @@ class TestResponseFailedSurfacing: def run_streaming(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]: del args - if kwargs.get("stream"): - return ResponseStream(_raise_stream(), finalizer=AgentResponse.from_updates) - raise NotImplementedError("Only streaming is configured on this mock") + assert kwargs.get("stream") is True + return ResponseStream(_raise_stream(), finalizer=AgentResponse.from_updates) agent.run = MagicMock(side_effect=run_streaming) server = _make_server(agent) @@ -4084,9 +4073,8 @@ class _ToolApprovalWorkflowAgentMock(SupportsAgentRun): **kwargs: Any, ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]: del session - if stream: - return self._run_stream(messages=messages, **kwargs) - return self._run(messages=messages, **kwargs) + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." + return self._run_stream(messages=messages, **kwargs) @staticmethod def _normalize( @@ -4114,20 +4102,6 @@ class _ToolApprovalWorkflowAgentMock(SupportsAgentRun): def _approval_responses_in(messages: list[Message]) -> list[Content]: return [c for m in messages for c in m.contents if c.type == "function_approval_response"] - async def _run( - self, - messages: str | Content | Message | Sequence[str | Content | Message] | None = None, - **kwargs: Any, - ) -> AgentResponse: - del kwargs - normalized = self._normalize(messages) - self.last_run_messages = normalized - self.run_count += 1 - if self._approval_responses_in(normalized): - return AgentResponse(messages=[Message("assistant", [Content.from_text(text=self._final_text)])]) - approval = self._build_approval_request() - return AgentResponse(messages=[Message("assistant", [approval])]) - def _run_stream( self, messages: str | Content | Message | Sequence[str | Content | Message] | None = None, @@ -4205,12 +4179,10 @@ def _build_text_workflow_agent(text: str) -> WorkflowAgent: **kwargs: Any, ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]: del messages, session, kwargs + assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents." text = self._text name = self.name - async def _aresult() -> AgentResponse: - return AgentResponse(messages=[Message("assistant", [Content.from_text(text=text)])]) - async def _aiter() -> AsyncIterator[AgentResponseUpdate]: yield AgentResponseUpdate( contents=[Content.from_text(text=text)], @@ -4218,9 +4190,7 @@ def _build_text_workflow_agent(text: str) -> WorkflowAgent: author_name=name, ) - if stream: - return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates) - return _aresult() + return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates) inner = _TextAgent("text-agent", text) diff --git a/python/packages/foundry_hosting/tests/test_responses_int.py b/python/packages/foundry_hosting/tests/test_responses_int.py index 1094d3e66..cac824afc 100644 --- a/python/packages/foundry_hosting/tests/test_responses_int.py +++ b/python/packages/foundry_hosting/tests/test_responses_int.py @@ -532,35 +532,58 @@ class TestReasoningHostedMcpReplay: "status": "completed", } + def _streaming_response(response_id: str, output: list[dict[str, Any]]) -> httpx.Response: + response = _response(response_id, output) + events: list[dict[str, Any]] = [] + for output_index, item in enumerate(output): + events.extend([ + { + "type": "response.output_item.added", + "output_index": output_index, + "item": {**item, "status": "in_progress"}, + "sequence_number": len(events), + }, + { + "type": "response.output_item.done", + "output_index": output_index, + "item": item, + "sequence_number": len(events) + 1, + }, + ]) + events.append({ + "type": "response.completed", + "response": response, + "sequence_number": len(events), + }) + body = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n" + return httpx.Response(200, text=body, headers={"content-type": "text/event-stream"}) + async def foundry_responses_boundary(request: httpx.Request) -> httpx.Response: nonlocal call_count call_count += 1 payload = json.loads(request.content) provider_payloads.append(payload) if call_count == 1: - return httpx.Response( - 200, - json=_response( - "resp_first", - [ - { - "encrypted_content": "encrypted-reasoning", - "id": reasoning_id, - "summary": [{"text": "The MCP server has the answer.", "type": "summary_text"}], - "type": "reasoning", - }, - { - "id": "mcp_paired", - "arguments": '{"query":"Agent Framework overview"}', - "name": "microsoft_docs_search", - "server_label": "Microsoft_Learn", - "type": "mcp_call", - "output": "Microsoft Agent Framework", - "status": "completed", - }, - _message("msg_first"), - ], - ), + return _streaming_response( + "resp_first", + [ + { + "encrypted_content": "encrypted-reasoning", + "id": reasoning_id, + "summary": [{"text": "The MCP server has the answer.", "type": "summary_text"}], + "type": "reasoning", + }, + { + "id": "mcp_paired", + "arguments": '{"query":"Agent Framework overview"}', + "name": "microsoft_docs_search", + "server_label": "Microsoft_Learn", + "type": "mcp_call", + "output": "Microsoft Agent Framework", + "status": "completed", + }, + _message("msg_first"), + ], ) input_items = payload["input"] @@ -586,13 +609,7 @@ class TestReasoningHostedMcpReplay: }, ) - return httpx.Response( - 200, - json=_response( - "resp_second", - [_message("msg_second")], - ), - ) + return _streaming_response("resp_second", [_message("msg_second")]) transport = httpx.MockTransport(foundry_responses_boundary) responses_client = AsyncOpenAI(