diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md index d08d0c162c5..f10b8b839df 100644 --- a/python/packages/hosting-responses/README.md +++ b/python/packages/hosting-responses/README.md @@ -6,8 +6,10 @@ This package provides the Responses-specific conversion layer: - `responses_to_run(...)` — convert a Responses request body into Agent Framework run values. -- `responses_session_id(...)` — extract a prior `resp_*` response id or - `conv_*` conversation id from the request body when present. +- `responses_session_id(...)` — return `(session_id, is_conversation_id)` for a + prior `resp_*` response id or `conv_*` conversation id, or `(None, None)` when + neither is present. +- `create_conversation_id(...)` — mint a Responses-shaped conversation id. - `create_response_id(...)` — mint a Responses-shaped response id. - `responses_from_run(...)` — convert an `AgentResponse` into a Responses-compatible JSON payload. @@ -35,7 +37,7 @@ state = AgentState(agent) @app.post("/responses") async def responses(body: dict = Body(...)) -> JSONResponse: run = responses_to_run(body) - session_id = responses_session_id(body) + session_id, is_conversation_id = responses_session_id(body) response_id = create_response_id() session = await state.get_or_create_session(session_id or response_id) result = await (await state.get_target()).run( @@ -43,12 +45,13 @@ async def responses(body: dict = Body(...)) -> JSONResponse: session=session, options=run["options"], ) - if body.get("conversation_id") == session_id: + if is_conversation_id: # The app must serialize writers that advance this stable id. await state.set_session(session_id, session) else: await state.set_session(response_id, session) - return JSONResponse(responses_from_run(result, response_id=response_id, session_id=session_id)) + conversation_id = session_id if is_conversation_id else None + return JSONResponse(responses_from_run(result, response_id=response_id, conversation_id=conversation_id)) ``` `previous_response_id` identifies an immutable continuation snapshot: multiple diff --git a/python/packages/hosting-responses/agent_framework_hosting_responses/__init__.py b/python/packages/hosting-responses/agent_framework_hosting_responses/__init__.py index 9fc6246abbf..eb32a76307f 100644 --- a/python/packages/hosting-responses/agent_framework_hosting_responses/__init__.py +++ b/python/packages/hosting-responses/agent_framework_hosting_responses/__init__.py @@ -5,6 +5,7 @@ import importlib.metadata from ._parsing import ( + create_conversation_id, create_response_id, messages_from_responses_input, responses_from_run, @@ -20,6 +21,7 @@ __all__ = [ "__version__", + "create_conversation_id", "create_response_id", "messages_from_responses_input", "responses_from_run", diff --git a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py index d8a5ecfaa9e..154ce300306 100644 --- a/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py +++ b/python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py @@ -16,6 +16,7 @@ import json import time import uuid +import warnings from collections.abc import AsyncIterator, Mapping, Sequence from typing import Any, cast @@ -125,26 +126,45 @@ def create_response_id() -> str: return f"resp_{uuid.uuid4().hex}" -def responses_session_id(body: Mapping[str, Any]) -> str | None: - """Return the Responses session id from request body, if present. +def create_conversation_id() -> str: + """Create a Responses-shaped conversation id.""" + return f"conv_{uuid.uuid4().hex}" - The returned value can be a ``resp_*`` previous response id or a ``conv_*`` - conversation id. Callers choose whether this request-derived value is + +def responses_session_id(body: Mapping[str, Any]) -> tuple[str, bool] | tuple[None, None]: + """Return the Responses session id and whether it is a conversation id. + + The session id can be a ``resp_*`` previous response id or a ``conv_*`` + conversation id. Callers choose whether these request-derived values are trusted for their route and deployment. Args: body: OpenAI Responses-shaped request body. Returns: - Previous response id, conversation id, or ``None``. + The session id, if present, and whether it came from ``conversation_id``. + The flag is ``None`` when no session id is present. """ previous_response_id = body.get("previous_response_id") - if isinstance(previous_response_id, str) and previous_response_id: - return previous_response_id + if isinstance(previous_response_id, str) and previous_response_id and not previous_response_id.startswith("resp_"): + warnings.warn( + "`previous_response_id` does not use the OpenAI Responses `resp_` prefix; " + "continuing with the supplied value.", + UserWarning, + stacklevel=2, + ) conversation_id = body.get("conversation_id") + if isinstance(conversation_id, str) and conversation_id and not conversation_id.startswith("conv_"): + warnings.warn( + "`conversation_id` does not use the OpenAI Responses `conv_` prefix; continuing with the supplied value.", + UserWarning, + stacklevel=2, + ) + if isinstance(previous_response_id, str) and previous_response_id: + return previous_response_id, False if isinstance(conversation_id, str) and conversation_id: - return conversation_id - return None + return conversation_id, True + return None, None def responses_to_run(body: Mapping[str, Any]) -> AgentRunArgs: @@ -176,7 +196,7 @@ def responses_from_run( result: AgentResponse[Any], *, response_id: str, - session_id: str | None = None, + conversation_id: str | None = None, ) -> dict[str, Any]: """Convert an Agent Framework response into a Responses payload. @@ -185,8 +205,7 @@ def responses_from_run( Keyword Args: response_id: Id for the response being created. - session_id: Optional prior ``resp_*`` or ``conv_*`` session id. When it - is a conversation id, the helper renders it in the Responses + conversation_id: Optional conversation id to render in the Responses conversation field. Returns: @@ -205,8 +224,8 @@ def responses_from_run( "tools": [], "metadata": {}, } - if session_id is not None and session_id.startswith("conv_"): - response_kwargs["conversation"] = {"id": session_id} + if conversation_id is not None: + response_kwargs["conversation"] = {"id": conversation_id} return _response_payload(OpenAIResponse(**response_kwargs)) @@ -839,7 +858,7 @@ async def responses_from_streaming_run( stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]], *, response_id: str, - session_id: str | None = None, + conversation_id: str | None = None, ) -> AsyncIterator[str]: """Convert an Agent Framework response stream into Responses SSE events. @@ -849,23 +868,26 @@ async def responses_from_streaming_run( Keyword Args: response_id: Id for the response being created. - session_id: Optional prior ``resp_*`` or ``conv_*`` session id. + conversation_id: Optional conversation id to render in Responses events. Yields: Server-Sent Event strings. """ + created_response: dict[str, Any] = { + "id": response_id, + "object": "response", + "created_at": int(time.time()), + "status": "in_progress", + "model": "agent", + "output": [], + } + if conversation_id is not None: + created_response["conversation"] = {"id": conversation_id} yield _sse_event( "response.created", { "type": "response.created", - "response": { - "id": response_id, - "object": "response", - "created_at": int(time.time()), - "status": "in_progress", - "model": "agent", - "output": [], - }, + "response": created_response, }, ) @@ -886,7 +908,7 @@ async def responses_from_streaming_run( ) final = await stream.get_final_response() - payload = responses_from_run(final, response_id=response_id, session_id=session_id) + payload = responses_from_run(final, response_id=response_id, conversation_id=conversation_id) if model is not None: # The finalized `AgentResponse` never carries a raw representation # (see `_model_from_update`), so prefer the model observed on the @@ -917,8 +939,8 @@ async def responses_from_streaming_run( "message": str(exc), }, } - if session_id is not None and session_id.startswith("conv_"): - response_kwargs["conversation"] = {"id": session_id} + if conversation_id is not None: + response_kwargs["conversation"] = {"id": conversation_id} yield _sse_event( "response.failed", { diff --git a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py index 7ede2c152e5..2772ca7b0c0 100644 --- a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py +++ b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py @@ -129,8 +129,8 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin run = responses_to_run(body) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - session_id = responses_session_id(body) - conversation_id = session_id if body.get("conversation_id") == session_id else None + session_id, is_conversation_id = responses_session_id(body) + conversation_id = session_id if is_conversation_id else None response_id = create_response_id() target = await state.get_target() @@ -146,7 +146,7 @@ async def stream_events() -> AsyncIterator[str]: async for event in responses_from_streaming_run( stream, response_id=response_id, - session_id=session_id, + conversation_id=conversation_id, ): yield event if conversation_id is not None: @@ -164,7 +164,7 @@ async def stream_events() -> AsyncIterator[str]: await state.set_session(conversation_id, session) else: await state.set_session(response_id, session) - return JSONResponse(responses_from_run(result, response_id=response_id, session_id=session_id)) + return JSONResponse(responses_from_run(result, response_id=response_id, conversation_id=conversation_id)) return app diff --git a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py index 072ac2c20c6..9377cc03eb1 100644 --- a/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py +++ b/python/packages/hosting-responses/tests/hosting_responses/test_parsing.py @@ -5,6 +5,7 @@ from __future__ import annotations import json +import warnings from collections.abc import AsyncIterator, Sequence from typing import cast @@ -12,6 +13,7 @@ from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message, ResponseStream from agent_framework_hosting_responses import ( + create_conversation_id, create_response_id, messages_from_responses_input, responses_from_run, @@ -118,19 +120,43 @@ def test_image_url_missing_raises(self) -> None: class TestResponsesRunHelpers: + def test_create_conversation_id_shape(self) -> None: + conversation_id = create_conversation_id() + + assert len(conversation_id) == 37 + assert conversation_id.startswith("conv_") + assert all(character in "0123456789abcdef" for character in conversation_id.removeprefix("conv_")) + def test_create_response_id_shape(self) -> None: response_id = create_response_id() assert response_id.startswith("resp_") def test_responses_session_id_prefers_previous_response(self) -> None: - assert responses_session_id({"previous_response_id": "resp_1", "conversation_id": "conv_1"}) == "resp_1" + assert responses_session_id({"previous_response_id": "resp_1", "conversation_id": "conv_1"}) == ( + "resp_1", + False, + ) + + def test_responses_session_id_valid_ids_do_not_warn(self) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert responses_session_id({"previous_response_id": "resp_1"}) == ("resp_1", False) + assert responses_session_id({"conversation_id": "conv_1"}) == ("conv_1", True) + + def test_responses_session_id_warns_for_nonstandard_previous_response_id(self) -> None: + with pytest.warns(UserWarning, match="previous_response_id.*resp_"): + assert responses_session_id({"previous_response_id": "custom-response"}) == ("custom-response", False) + + def test_responses_session_id_warns_for_nonstandard_conversation_id(self) -> None: + with pytest.warns(UserWarning, match="conversation_id.*conv_"): + assert responses_session_id({"conversation_id": "custom-conversation"}) == ("custom-conversation", True) def test_responses_session_id_uses_conversation_id(self) -> None: - assert responses_session_id({"conversation_id": "conv_1"}) == "conv_1" + assert responses_session_id({"conversation_id": "conv_1"}) == ("conv_1", True) def test_responses_session_id_returns_none_when_absent(self) -> None: - assert responses_session_id({"input": "hi"}) is None + assert responses_session_id({"input": "hi"}) == (None, None) def test_responses_to_run_returns_messages_options_and_stream(self) -> None: run = responses_to_run({ @@ -200,17 +226,17 @@ def test_responses_from_run_preserves_multimodal_output_items(self) -> None: ] assert output[3]["content"][0]["text"] == "done" - def test_responses_from_run_maps_conversation_session(self) -> None: + def test_responses_from_run_maps_conversation_id(self) -> None: result = AgentResponse(messages=Message(role="assistant", contents=[Content.from_text("hello")])) - payload = responses_from_run(result, response_id="resp_new", session_id="conv_1") + payload = responses_from_run(result, response_id="resp_new", conversation_id="conv_1") assert payload["conversation"] == {"id": "conv_1"} - def test_responses_from_run_omits_previous_response_session(self) -> None: + def test_responses_from_run_omits_conversation_when_absent(self) -> None: result = AgentResponse(messages=Message(role="assistant", contents=[Content.from_text("hello")])) - payload = responses_from_run(result, response_id="resp_new", session_id="resp_1") + payload = responses_from_run(result, response_id="resp_new") assert "conversation" not in payload @@ -229,11 +255,12 @@ def finalizer(items: Sequence[AgentResponseUpdate]) -> AgentResponse: async for event in responses_from_streaming_run( stream, response_id="resp_new", - session_id="conv_1", + conversation_id="conv_1", ) ] assert events[0].startswith("event: response.created") + assert '"conversation":{"id":"conv_1"}' in events[0] assert "response.output_text.delta" in events[1] assert "hel" in events[1] assert "lo" in events[2] @@ -252,7 +279,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]: async for event in responses_from_streaming_run( stream, response_id="resp_new", - session_id="conv_1", + conversation_id="conv_1", ) ] diff --git a/python/samples/04-hosting/af-hosting/local_responses/app.py b/python/samples/04-hosting/af-hosting/local_responses/app.py index 88c22315c71..0b3503b0262 100644 --- a/python/samples/04-hosting/af-hosting/local_responses/app.py +++ b/python/samples/04-hosting/af-hosting/local_responses/app.py @@ -117,8 +117,8 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin run = responses_to_run(body) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - session_id = responses_session_id(body) - conversation_id = session_id if body.get("conversation_id") == session_id else None + session_id, is_conversation_id = responses_session_id(body) + conversation_id = session_id if is_conversation_id else None response_id = create_response_id() # App-specific policy: allow only the request options this route is willing @@ -148,7 +148,7 @@ async def stream_events() -> AsyncIterator[str]: async for event in responses_from_streaming_run( stream, response_id=response_id, - session_id=session_id, + conversation_id=conversation_id, ): yield event # `agent.run(..., stream=True)` updates the session while the stream @@ -184,7 +184,7 @@ async def stream_events() -> AsyncIterator[str]: responses_from_run( result, response_id=response_id, - session_id=session_id, + conversation_id=conversation_id, ) ) diff --git a/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py b/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py index 4d779d66db8..dc49915d9c1 100644 --- a/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py +++ b/python/samples/04-hosting/af-hosting/local_responses_workflow/app.py @@ -230,10 +230,10 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse: # noqa: raise HTTPException(status_code=400, detail=str(exc)) from exc # This sample demonstrates only Responses `previous_response_id` - # continuation. `responses_session_id` also returns `conversation_id`, so - # reject that shape here instead of treating it as a checkpoint cursor. - previous_response_id = responses_session_id(body) - if previous_response_id and not previous_response_id.startswith("resp_"): + # continuation, so reject `conversation_id` instead of treating it as a + # checkpoint cursor. + previous_response_id, is_conversation_id = responses_session_id(body) + if is_conversation_id: raise HTTPException( status_code=400, detail="This server supports previous_response_id continuation only; conversation_id is not implemented.", @@ -267,7 +267,6 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse: # noqa: responses_from_run( response_from_workflow_result(result), response_id=response_id, - session_id=previous_response_id, ) )