mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: add agent-framework-hosting-responses channel (#5639)
* feat(hosting-responses): add OpenAI Responses-shaped channel package
New ``agent-framework-hosting-responses`` package implementing the
OpenAI Responses-shaped HTTP channel for the Hosting framework. Mounts
``POST /responses`` (and a ``/responses/{response_id}`` GET) onto an
``AgentFrameworkHost`` and translates the OpenAI Responses wire shape
to/from the channel-neutral ``ChannelRequest`` / ``HostedRunResult``
plumbing.
Surface (re-exported from ``agent_framework_hosting_responses``):
- ``ResponsesChannel`` -- concrete ``Channel`` implementation. Owns the
Starlette route(s), parses inbound JSON into ``ChannelRequest``, runs
the optional ``ChannelRunHook``, calls back into the
``ChannelContext`` to invoke the agent target, builds Responses
envelopes (sync JSON or SSE), and respects
``DeliveryReport.include_originating`` so cross-channel push routes
only ack to the originating Responses caller.
- The minted ``response_id`` is propagated via the host's ContextVar
machinery so storage-side history providers (e.g.
``FoundryHostedAgentHistoryProvider``) persist envelopes against the
same id the channel returns.
- 48 unit tests covering route wiring, parsing of each Responses input
shape, hook composition, sync vs streaming paths, and originating
vs non-originating delivery branches.
Registers the package in ``python/pyproject.toml`` ``[tool.uv.sources]``
and adds the matching pyright ``executionEnvironments`` entry.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* review: address PR-3 round 2 feedback
- consume IsolationKeys.chat_key from the host-bound contextvar instead
of the raw `x-agent-chat-isolation-key` header off the wire so the
host's ASGI isolation middleware (or any operator-supplied
replacement) is the authoritative point at which the caller is
authenticated and the bucket key is established
- expand `response_id_factory` docstring to call out partition
co-location vs. partition-ownership enforcement: the channel forwards
`previous_response_id` as a hint to the factory; the storage layer
validates the embedded partition against the bound user/chat
isolation keys
- on mid-stream failure, call `deliver_response` with the accumulated
text before emitting `response.failed` so host-side history /
push-channel state stays consistent with the partial deltas the
client already saw
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* docs(hosting-responses): fix quickstart to use current Agent API
ChatAgent was renamed to Agent and ChatMessage to Message. Update the
README quickstart to use client.as_agent(...) and refresh the stale
docstring reference in _channel.py.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* fix(hosting-responses): adapt to hosted run result wrapper
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* feat(hosting-responses): add response hooks
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* fix(hosting-responses): keep instructions in chat options
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---------
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
committed by
GitHub
Unverified
GitHub
parent
4c317eb7cf
commit
d75f55ee2c
@@ -0,0 +1,27 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""OpenAI Responses-shaped channel for ``agent-framework-hosting``."""
|
||||
|
||||
import importlib.metadata
|
||||
|
||||
from ._channel import ResponsesChannel
|
||||
from ._parsing import (
|
||||
messages_from_responses_input,
|
||||
parse_response_target,
|
||||
parse_responses_identity,
|
||||
parse_responses_request,
|
||||
)
|
||||
|
||||
try:
|
||||
__version__ = importlib.metadata.version(__name__)
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
__version__ = "0.0.0"
|
||||
|
||||
__all__ = [
|
||||
"ResponsesChannel",
|
||||
"__version__",
|
||||
"messages_from_responses_input",
|
||||
"parse_response_target",
|
||||
"parse_responses_identity",
|
||||
"parse_responses_request",
|
||||
]
|
||||
@@ -0,0 +1,427 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""``ResponsesChannel`` — OpenAI Responses-shaped HTTP surface.
|
||||
|
||||
Exposes a single ``POST /responses`` endpoint that accepts
|
||||
``{"input": "...", "stream": false}`` (and the rest of the Responses API
|
||||
request body) and returns either a Responses-shaped JSON body
|
||||
(``stream=False``, default) or a Server-Sent-Events stream
|
||||
(``stream=True``).
|
||||
|
||||
Payload construction reuses the ``openai.types.responses`` Pydantic
|
||||
models so the OpenAI Python SDK ``stream=True`` consumer parses every
|
||||
required field without surprises.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework import AgentResponse, Content, Message
|
||||
from agent_framework_hosting import (
|
||||
ChannelContext,
|
||||
ChannelContribution,
|
||||
ChannelRequest,
|
||||
ChannelResponseContext,
|
||||
ChannelResponseHook,
|
||||
ChannelRunHook,
|
||||
ChannelSession,
|
||||
ChannelStreamTransformHook,
|
||||
HostedRunResult,
|
||||
apply_response_hook,
|
||||
apply_run_hook,
|
||||
get_current_isolation_keys,
|
||||
logger,
|
||||
)
|
||||
from openai.types.responses import (
|
||||
Response as OpenAIResponse,
|
||||
)
|
||||
from openai.types.responses import (
|
||||
ResponseCompletedEvent,
|
||||
ResponseCreatedEvent,
|
||||
ResponseError,
|
||||
ResponseFailedEvent,
|
||||
ResponseOutputMessage,
|
||||
ResponseOutputText,
|
||||
ResponseTextDeltaEvent,
|
||||
)
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse, Response, StreamingResponse
|
||||
from starlette.routing import Route
|
||||
|
||||
from ._parsing import (
|
||||
parse_response_target,
|
||||
parse_responses_identity,
|
||||
parse_responses_request,
|
||||
)
|
||||
|
||||
|
||||
def _ack_text() -> str:
|
||||
"""Tiny acknowledgement string for the originating wire.
|
||||
|
||||
Used when the agent reply is delivered out-of-band via :class:`ChannelPush`.
|
||||
"""
|
||||
return "[delivered out-of-band]"
|
||||
|
||||
|
||||
def _text_result(text: str) -> HostedRunResult[AgentResponse]:
|
||||
"""Build a host delivery payload from text accumulated by this channel."""
|
||||
return HostedRunResult(AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text(text=text)])]))
|
||||
|
||||
|
||||
class ResponsesChannel:
|
||||
"""Minimal OpenAI-Responses-shaped surface.
|
||||
|
||||
Mounts ``POST <path>/responses`` (default path ``/responses`` so the
|
||||
full route is ``/responses/responses`` when the channel is prefixed,
|
||||
or just ``/responses`` when ``path=""``).
|
||||
"""
|
||||
|
||||
name = "responses"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
path: str = "",
|
||||
run_hook: ChannelRunHook | None = None,
|
||||
response_hook: ChannelResponseHook | None = None,
|
||||
stream_transform_hook: ChannelStreamTransformHook | None = None,
|
||||
response_id_factory: Callable[..., str] | None = None,
|
||||
) -> None:
|
||||
"""Create a Responses channel.
|
||||
|
||||
Keyword Args:
|
||||
path: Mount prefix on the host. Default ``""`` mounts the
|
||||
``POST /responses`` route at the app root, matching the
|
||||
upstream OpenAI surface.
|
||||
run_hook: Optional :data:`ChannelRunHook` invoked with the
|
||||
parsed :class:`ChannelRequest` before the agent target
|
||||
runs. May return a replacement request.
|
||||
response_hook: Optional :data:`ChannelResponseHook` invoked
|
||||
before the channel serializes an originating
|
||||
:class:`HostedRunResult` into a Responses envelope. The
|
||||
host also invokes this hook when delivering to this
|
||||
channel as a non-originating push destination.
|
||||
stream_transform_hook: Optional per-update transform hook
|
||||
applied while streaming Server-Sent Events. Return a
|
||||
replacement update, or ``None`` to drop the update.
|
||||
response_id_factory: Optional callable that mints the
|
||||
per-request response id. Default produces
|
||||
``resp_<uuid hex>`` which matches the OpenAI Responses
|
||||
wire shape. Override when the host backing storage
|
||||
requires a different id format (e.g. Foundry storage,
|
||||
whose partition keys are encoded in the id and which
|
||||
rejects free-form ``resp_*`` ids with a server error).
|
||||
The same id is used for the channel envelope and for
|
||||
the host-side anchoring (``ChannelRequest.attributes``)
|
||||
so storage and replay agree.
|
||||
|
||||
Security note on partition co-location: when a caller
|
||||
supplies ``previous_response_id`` we forward it to the
|
||||
factory so id backends that embed partition keys can
|
||||
co-locate the new record with the chain's existing
|
||||
partition. The factory passes that hint through to the
|
||||
storage layer; **partition ownership is enforced at
|
||||
the storage layer**, not in the channel: the Foundry
|
||||
storage provider, for example, validates the request
|
||||
against the bound user/chat isolation keys and rejects
|
||||
writes whose embedded partition does not match the
|
||||
authenticated caller's isolation. Channel-level
|
||||
forwarding is therefore a performance hint, not a
|
||||
security boundary; the host's isolation middleware
|
||||
must establish the caller's identity before this
|
||||
route is entered.
|
||||
"""
|
||||
self.path = path
|
||||
self._hook = run_hook
|
||||
self.response_hook = response_hook
|
||||
self._stream_transform_hook = stream_transform_hook
|
||||
self._ctx: ChannelContext | None = None
|
||||
self._response_id_factory: Callable[..., str] = (
|
||||
response_id_factory if response_id_factory is not None else (lambda *_a, **_kw: f"resp_{uuid.uuid4().hex}")
|
||||
)
|
||||
|
||||
def contribute(self, context: ChannelContext) -> ChannelContribution:
|
||||
"""Capture the host-supplied context and register ``POST /responses``."""
|
||||
self._ctx = context
|
||||
return ChannelContribution(routes=[Route("/responses", self._handle, methods=["POST"])])
|
||||
|
||||
async def _handle(self, request: Request) -> Response:
|
||||
"""Handle a single ``POST /responses`` call.
|
||||
|
||||
Parses the OpenAI Responses-shaped body into ``Message`` /
|
||||
``options`` / ``ChannelSession`` triples via :mod:`._parsing`,
|
||||
applies the optional ``run_hook``, and either streams an SSE
|
||||
response stream or returns a one-shot OpenAI ``Response`` envelope.
|
||||
Non-originating ``response_target`` values resolve to a delivery
|
||||
acknowledgement instead of echoing the agent text on this wire.
|
||||
"""
|
||||
if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle
|
||||
return JSONResponse({"error": "channel not initialized"}, status_code=500)
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
return JSONResponse({"error": "invalid json"}, status_code=400)
|
||||
|
||||
try:
|
||||
messages, options, session = parse_responses_request(body)
|
||||
except ValueError as exc:
|
||||
return JSONResponse({"error": str(exc)}, status_code=422)
|
||||
|
||||
# When no ``previous_response_id`` chain anchor is on the body,
|
||||
# surface the isolation key the **host** lifted off the request
|
||||
# (via ``_FoundryIsolationASGIMiddleware`` for the default
|
||||
# Foundry-platform deployment, or whatever middleware the
|
||||
# operator configured in front of the host) as the channel
|
||||
# session id, so callers without an explicit anchor still get
|
||||
# a stable per-conversation session id (used by non-Foundry
|
||||
# history providers, routing/idempotency, etc.).
|
||||
#
|
||||
# Security note: we consume the host-bound contextvar set by the
|
||||
# ASGI isolation middleware, NOT the raw header off the wire.
|
||||
# That middleware is the operator's place to enforce auth and
|
||||
# gate which callers get to set isolation. If you mount the host
|
||||
# in front of a custom auth boundary, your middleware should
|
||||
# validate the caller before stamping ``set_current_isolation_keys``;
|
||||
# never trust raw wire headers to identify a session bucket.
|
||||
# The chat-iso value is *not* a valid storage anchor: the
|
||||
# Foundry history provider deliberately ignores it — multi-turn
|
||||
# storage chaining goes through the ``previous_response_id`` /
|
||||
# bound ``response_id`` pair on ``ChannelRequest.attributes``.
|
||||
bound_keys = get_current_isolation_keys()
|
||||
chat_iso = bound_keys.chat_key if bound_keys is not None else None
|
||||
if session is None and chat_iso:
|
||||
session = ChannelSession(isolation_key=chat_iso)
|
||||
|
||||
# Mint the response id once per request so the channel envelope
|
||||
# (one-shot or streamed) and any host-side anchoring (e.g. the
|
||||
# Foundry history provider's ``bind_request_context``) agree on
|
||||
# the same handle. The next turn arrives with this value as
|
||||
# ``previous_response_id`` and the storage chain walks. We pass
|
||||
# both anchors via ``ChannelRequest.attributes`` so the host
|
||||
# can pick them up without a channel-specific contract.
|
||||
previous_response_id: str | None = None
|
||||
prev_raw = body.get("previous_response_id")
|
||||
if isinstance(prev_raw, str) and prev_raw:
|
||||
previous_response_id = prev_raw
|
||||
# Pass the previous id (if any) as a hint to the factory so id
|
||||
# backends that embed partition keys (e.g. Foundry storage) can
|
||||
# co-locate the new record with the chain's existing partition.
|
||||
# No-arg factories continue to work via ``Callable[..., str]``.
|
||||
response_id = self._response_id_factory(previous_response_id)
|
||||
|
||||
attributes: dict[str, Any] = {"response_id": response_id}
|
||||
if previous_response_id is not None:
|
||||
attributes["previous_response_id"] = previous_response_id
|
||||
|
||||
# Honor the OpenAI-Responses ``stream`` flag — non-streaming by
|
||||
# default, SSE when the caller opts in. Run hooks may still flip
|
||||
# this per-request (e.g. force non-streaming for a particular user).
|
||||
channel_request = ChannelRequest(
|
||||
channel=self.name,
|
||||
operation="message.create",
|
||||
input=messages,
|
||||
session=session,
|
||||
options=options or None,
|
||||
stream=bool(body.get("stream", False)),
|
||||
identity=parse_responses_identity(body, self.name),
|
||||
response_target=parse_response_target(body),
|
||||
attributes=attributes,
|
||||
)
|
||||
|
||||
if self._hook is not None:
|
||||
channel_request = await apply_run_hook(
|
||||
self._hook,
|
||||
channel_request,
|
||||
target=self._ctx.target,
|
||||
protocol_request=body,
|
||||
)
|
||||
|
||||
if channel_request.stream:
|
||||
return StreamingResponse(
|
||||
self._stream_events(channel_request, body, response_id=response_id),
|
||||
media_type="text/event-stream",
|
||||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
||||
)
|
||||
|
||||
result = await self._ctx.run(channel_request)
|
||||
include_originating = await self._ctx.deliver_response(channel_request, result)
|
||||
if include_originating:
|
||||
result = await self._apply_response_hook(result, channel_request)
|
||||
text = result.result.text if include_originating else _ack_text()
|
||||
envelope = self._build_response(body, text, status="completed", response_id=response_id)
|
||||
return JSONResponse(envelope.model_dump(mode="json", exclude_none=True))
|
||||
|
||||
async def _apply_response_hook(
|
||||
self,
|
||||
result: HostedRunResult[AgentResponse],
|
||||
request: ChannelRequest,
|
||||
) -> HostedRunResult[AgentResponse]:
|
||||
"""Apply the channel-level response hook for an originating reply."""
|
||||
if self.response_hook is None:
|
||||
return result
|
||||
context = ChannelResponseContext(
|
||||
request=request,
|
||||
channel_name=self.name,
|
||||
destination_identity=None,
|
||||
originating=True,
|
||||
is_echo=False,
|
||||
)
|
||||
shaped = await apply_response_hook(self.response_hook, result, context=context)
|
||||
return cast("HostedRunResult[AgentResponse]", shaped)
|
||||
|
||||
def _build_response(
|
||||
self,
|
||||
body: Mapping[str, Any],
|
||||
text: str,
|
||||
*,
|
||||
status: str,
|
||||
response_id: str | None = None,
|
||||
) -> OpenAIResponse:
|
||||
"""Construct an OpenAI ``Response`` for a finished (non-streaming) run.
|
||||
|
||||
``status`` mirrors the top-level Response status set values
|
||||
(``in_progress`` / ``completed`` / ``failed`` / ``incomplete`` /
|
||||
``cancelled``). The nested ``ResponseOutputMessage.status`` field
|
||||
only accepts ``in_progress`` / ``completed`` / ``incomplete``, so
|
||||
terminal-but-non-success states collapse to ``incomplete`` there
|
||||
— the failure detail still travels via the top-level ``status``
|
||||
and (for streamed errors) the ``error`` field.
|
||||
|
||||
``response_id``: the per-request id minted in :meth:`_handle`.
|
||||
Passed in so envelope and storage agree on a single handle per
|
||||
turn (see :meth:`_handle` notes). Falls back to a fresh uuid
|
||||
when callers (e.g. :meth:`_stream_events`'s skeleton path
|
||||
before this argument was introduced) don't supply one.
|
||||
"""
|
||||
message_status = status if status in ("in_progress", "completed", "incomplete") else "incomplete"
|
||||
return OpenAIResponse(
|
||||
id=response_id or self._response_id_factory(None),
|
||||
object="response",
|
||||
created_at=time.time(),
|
||||
status=status, # type: ignore[arg-type]
|
||||
model=body.get("model", "agent"),
|
||||
output=[
|
||||
ResponseOutputMessage(
|
||||
id=f"msg_{uuid.uuid4().hex}",
|
||||
type="message",
|
||||
role="assistant",
|
||||
status=message_status, # type: ignore[arg-type]
|
||||
content=[ResponseOutputText(type="output_text", text=text, annotations=[])],
|
||||
)
|
||||
],
|
||||
parallel_tool_calls=False,
|
||||
tool_choice="auto",
|
||||
tools=[],
|
||||
metadata={},
|
||||
)
|
||||
|
||||
async def _stream_events(
|
||||
self,
|
||||
request: ChannelRequest,
|
||||
body: Mapping[str, Any],
|
||||
*,
|
||||
response_id: str,
|
||||
) -> AsyncIterator[str]:
|
||||
"""Yield SSE events shaped like the OpenAI Responses streaming protocol.
|
||||
|
||||
Emits ``response.created`` → many ``response.output_text.delta``
|
||||
→ ``response.completed`` (or ``response.failed`` on error).
|
||||
"""
|
||||
if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle
|
||||
return
|
||||
|
||||
msg_id = f"msg_{uuid.uuid4().hex}"
|
||||
seq = 0
|
||||
|
||||
def next_seq() -> int:
|
||||
nonlocal seq
|
||||
seq += 1
|
||||
return seq
|
||||
|
||||
def sse(event: Any) -> str:
|
||||
return f"event: {event.type}\ndata: {event.model_dump_json(exclude_none=True)}\n\n"
|
||||
|
||||
skeleton = self._build_response(body, "", status="in_progress", response_id=response_id)
|
||||
yield sse(ResponseCreatedEvent(type="response.created", response=skeleton, sequence_number=next_seq()))
|
||||
|
||||
accumulated = ""
|
||||
try:
|
||||
stream = self._ctx.run_stream(request)
|
||||
async for update in stream:
|
||||
if self._stream_transform_hook is not None:
|
||||
transformed = self._stream_transform_hook(update)
|
||||
update = await transformed if isinstance(transformed, Awaitable) else transformed
|
||||
if update is None:
|
||||
continue
|
||||
chunk = getattr(update, "text", None)
|
||||
if chunk:
|
||||
accumulated += chunk
|
||||
yield sse(
|
||||
ResponseTextDeltaEvent(
|
||||
type="response.output_text.delta",
|
||||
item_id=msg_id,
|
||||
output_index=0,
|
||||
content_index=0,
|
||||
delta=chunk,
|
||||
logprobs=[],
|
||||
sequence_number=next_seq(),
|
||||
)
|
||||
)
|
||||
try:
|
||||
# Finalize so context-provider / history hooks on the agent
|
||||
# still run even though we are emitting our own SSE.
|
||||
await stream.get_final_response()
|
||||
except Exception: # pragma: no cover - finalize is best-effort
|
||||
logger.exception("Responses stream finalize failed")
|
||||
except Exception as exc:
|
||||
logger.exception("Responses stream consumption failed")
|
||||
# Mid-stream failure: the wire already saw partial deltas
|
||||
# so host-side state must reflect that — call
|
||||
# ``deliver_response`` with the accumulated text (best-effort)
|
||||
# before signalling failure to the client. Without this,
|
||||
# next turn's chain anchored on this ``response_id`` would
|
||||
# be inconsistent with what the user actually saw, and any
|
||||
# non-originating push targets would silently miss the turn.
|
||||
# ``deliver_response`` itself is best-effort; we swallow its
|
||||
# exceptions so the failure event still reaches the client.
|
||||
try:
|
||||
await self._ctx.deliver_response(request, _text_result(accumulated))
|
||||
except Exception: # pragma: no cover - delivery is best-effort
|
||||
logger.exception("Responses stream failure deliver_response failed")
|
||||
failed = self._build_response(body, accumulated, status="failed", response_id=response_id)
|
||||
failed.error = ResponseError(code="server_error", message=str(exc))
|
||||
yield sse(
|
||||
ResponseFailedEvent(
|
||||
type="response.failed",
|
||||
response=failed,
|
||||
sequence_number=next_seq(),
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
completed_text = accumulated
|
||||
result = _text_result(accumulated)
|
||||
include_originating = await self._ctx.deliver_response(request, result)
|
||||
if include_originating:
|
||||
result = await self._apply_response_hook(result, request)
|
||||
completed_text = result.result.text
|
||||
else:
|
||||
completed_text = _ack_text()
|
||||
completed = self._build_response(body, completed_text, status="completed", response_id=response_id)
|
||||
# Reuse the same message id we emitted deltas under.
|
||||
if completed.output and isinstance(completed.output[0], ResponseOutputMessage):
|
||||
completed.output[0].id = msg_id
|
||||
yield sse(
|
||||
ResponseCompletedEvent(
|
||||
type="response.completed",
|
||||
response=completed,
|
||||
sequence_number=next_seq(),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["ResponsesChannel"]
|
||||
@@ -0,0 +1,234 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Parsing helpers for the OpenAI Responses-API request body.
|
||||
|
||||
The Responses API accepts ``input`` as either a string or a list of "input
|
||||
items". An item is either a content part (``input_text`` / ``input_image``
|
||||
/ ``input_file``) or a message envelope ``{type: "message", role,
|
||||
content: [...]}``. We translate that into an Agent Framework ``Message``
|
||||
list and split out the ChatOptions-shaped fields the API also carries.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework import Content, Message
|
||||
from agent_framework_hosting import ChannelIdentity, ChannelSession, ResponseTarget, logger
|
||||
|
||||
# OpenAI Responses field name → Agent Framework ChatOptions field name.
|
||||
_RESPONSES_OPTION_REMAP = {
|
||||
"max_output_tokens": "max_tokens",
|
||||
"parallel_tool_calls": "allow_multiple_tool_calls",
|
||||
}
|
||||
# Fields we forward to ChatOptions verbatim. ``instructions`` stays here
|
||||
# because Agent Framework exposes it as a ChatOptions field; it must not be
|
||||
# lifted into a synthetic system message.
|
||||
_RESPONSES_OPTION_PASSTHROUGH = {
|
||||
"instructions",
|
||||
"temperature",
|
||||
"top_p",
|
||||
"metadata",
|
||||
"user",
|
||||
"safety_identifier",
|
||||
"tool_choice",
|
||||
"tools",
|
||||
"store",
|
||||
"response_format",
|
||||
"stop",
|
||||
"seed",
|
||||
"frequency_penalty",
|
||||
"presence_penalty",
|
||||
"logit_bias",
|
||||
}
|
||||
# Fields the Responses transport owns; they must not be forwarded as options.
|
||||
_RESPONSES_TRANSPORT_KEYS = {"input", "model", "stream", "previous_response_id", "response_target"}
|
||||
|
||||
|
||||
def parse_response_target(body: Mapping[str, Any]) -> ResponseTarget:
|
||||
"""Translate the OpenAI Responses ``response_target`` field into a :class:`ResponseTarget`.
|
||||
|
||||
Accepted shapes:
|
||||
|
||||
- ``"originating"`` / ``"active"`` / ``"all_linked"`` / ``"none"`` — bare strings.
|
||||
- ``"telegram"`` / ``"telegram:<chat_id>"`` — single channel destination.
|
||||
- ``["telegram:<id>", "originating"]`` — list of destinations; the
|
||||
pseudo-name ``"originating"`` includes the originating channel.
|
||||
- ``{"channels": [...]}`` — same list semantics with the explicit key.
|
||||
- ``{"kind": "active"}`` / ``{"kind": "all_linked"}`` — explicit kind.
|
||||
|
||||
Anything malformed is logged at WARNING and falls back to ``originating``.
|
||||
"""
|
||||
raw = body.get("response_target")
|
||||
if raw is None:
|
||||
return ResponseTarget.originating # type: ignore[attr-defined,no-any-return]
|
||||
if isinstance(raw, str):
|
||||
keyword = raw.strip()
|
||||
if keyword == "originating":
|
||||
return ResponseTarget.originating # type: ignore[attr-defined,no-any-return]
|
||||
if keyword == "active":
|
||||
return ResponseTarget.active # type: ignore[attr-defined,no-any-return]
|
||||
if keyword == "all_linked":
|
||||
return ResponseTarget.all_linked # type: ignore[attr-defined,no-any-return]
|
||||
if keyword == "none":
|
||||
return ResponseTarget.none # type: ignore[attr-defined,no-any-return]
|
||||
# Treat any other bare string as a single channel destination.
|
||||
return ResponseTarget.channel(keyword)
|
||||
if isinstance(raw, list):
|
||||
return _parse_channels_list(cast("list[Any]", raw)) # type: ignore[redundant-cast]
|
||||
if isinstance(raw, Mapping):
|
||||
raw_map = cast("Mapping[str, Any]", raw)
|
||||
channels = raw_map.get("channels")
|
||||
if isinstance(channels, list):
|
||||
return _parse_channels_list(cast("list[Any]", channels)) # type: ignore[redundant-cast]
|
||||
kind = raw_map.get("kind")
|
||||
if kind == "active":
|
||||
return ResponseTarget.active # type: ignore[attr-defined,no-any-return]
|
||||
if kind == "all_linked":
|
||||
return ResponseTarget.all_linked # type: ignore[attr-defined,no-any-return]
|
||||
if kind == "none":
|
||||
return ResponseTarget.none # type: ignore[attr-defined,no-any-return]
|
||||
if kind == "originating":
|
||||
return ResponseTarget.originating # type: ignore[attr-defined,no-any-return]
|
||||
logger.warning("responses: ignoring malformed response_target=%r", cast("Any", raw))
|
||||
return ResponseTarget.originating # type: ignore[attr-defined,no-any-return]
|
||||
|
||||
|
||||
def _parse_channels_list(raw: list[Any]) -> ResponseTarget:
|
||||
"""Build a ``ResponseTarget.channels`` from a raw list, dropping non-string entries.
|
||||
|
||||
An empty list (or one with no usable strings) collapses back to
|
||||
``originating`` so we never silently produce a target that nobody
|
||||
will deliver to.
|
||||
"""
|
||||
tokens = [t for t in raw if isinstance(t, str) and t]
|
||||
if len(tokens) != len(raw):
|
||||
logger.warning("responses: dropping non-string entries from response_target=%r", raw)
|
||||
if not tokens:
|
||||
return ResponseTarget.originating # type: ignore[attr-defined,no-any-return]
|
||||
return ResponseTarget.channels(tokens)
|
||||
|
||||
|
||||
def parse_responses_identity(body: Mapping[str, Any], channel_name: str) -> ChannelIdentity | None:
|
||||
"""Surface the caller as a :class:`ChannelIdentity` so the host can record it.
|
||||
|
||||
OpenAI Responses replaced ``user`` with ``safety_identifier`` — we use
|
||||
that as the native id, falling back to the legacy ``user`` field.
|
||||
"""
|
||||
native = body.get("safety_identifier") or body.get("user")
|
||||
if not isinstance(native, str) or not native:
|
||||
return None
|
||||
return ChannelIdentity(channel=channel_name, native_id=native)
|
||||
|
||||
|
||||
def _content_from_input_item(item: Mapping[str, Any]) -> Content:
|
||||
"""Convert a single OpenAI Responses ``input`` item into a :class:`Content` part.
|
||||
|
||||
Handles the ``input_text``/``output_text``/``text`` text variants,
|
||||
``input_image`` URL references, and ``input_file`` references via either
|
||||
a public URL or a hosted ``file_id``. Raises ``ValueError`` for any
|
||||
unsupported item type so the surrounding parser can return a 422.
|
||||
"""
|
||||
item_type = item.get("type")
|
||||
if item_type in ("input_text", "output_text", "text"):
|
||||
return Content.from_text(text=str(item.get("text", "")))
|
||||
if item_type == "input_image":
|
||||
image_url: Any = item.get("image_url")
|
||||
if isinstance(image_url, Mapping):
|
||||
image_url = cast("Mapping[str, Any]", image_url).get("url")
|
||||
if not isinstance(image_url, str):
|
||||
raise ValueError("input_image requires `image_url`")
|
||||
return Content.from_uri(uri=image_url, media_type="image/*")
|
||||
if item_type == "input_file":
|
||||
if (uri := item.get("file_url")) and isinstance(uri, str):
|
||||
return Content.from_uri(uri=uri, media_type=item.get("mime_type"))
|
||||
if file_id := item.get("file_id"):
|
||||
return Content(type="hosted_file", file_id=str(file_id))
|
||||
raise ValueError("input_file requires `file_url` or `file_id`")
|
||||
raise ValueError(f"Unsupported Responses input content type: {item_type!r}")
|
||||
|
||||
|
||||
def messages_from_responses_input(value: Any) -> list[Message]:
|
||||
"""Translate ``input`` (string or list of items) into :class:`Message` objects."""
|
||||
if isinstance(value, str):
|
||||
return [Message("user", [Content.from_text(text=value)])]
|
||||
if not isinstance(value, list) or not value:
|
||||
raise ValueError("`input` must be a non-empty string or list")
|
||||
|
||||
messages: list[Message] = []
|
||||
pending_user_parts: list[Content] = []
|
||||
|
||||
def flush() -> None:
|
||||
"""Emit any buffered loose user content as a single user message."""
|
||||
if pending_user_parts:
|
||||
messages.append(Message("user", list(pending_user_parts)))
|
||||
pending_user_parts.clear()
|
||||
|
||||
for item in cast("list[Any]", value): # type: ignore[redundant-cast]
|
||||
if not isinstance(item, Mapping):
|
||||
raise ValueError("each `input` item must be an object")
|
||||
item_map = cast("Mapping[str, Any]", item)
|
||||
if item_map.get("type") == "message":
|
||||
flush()
|
||||
role = str(item_map.get("role") or "user")
|
||||
content: Any = item_map.get("content") or []
|
||||
parts: list[Content]
|
||||
if isinstance(content, str):
|
||||
parts = [Content.from_text(text=content)]
|
||||
elif isinstance(content, list):
|
||||
parts = [
|
||||
_content_from_input_item(cast("Mapping[str, Any]", c))
|
||||
for c in cast("list[Any]", content) # type: ignore[redundant-cast]
|
||||
if isinstance(c, Mapping)
|
||||
]
|
||||
else:
|
||||
parts = []
|
||||
messages.append(Message(role, parts))
|
||||
else:
|
||||
pending_user_parts.append(_content_from_input_item(item_map))
|
||||
|
||||
flush()
|
||||
if not messages:
|
||||
raise ValueError("`input` produced no messages")
|
||||
return messages
|
||||
|
||||
|
||||
def parse_responses_request(
|
||||
body: Mapping[str, Any],
|
||||
) -> tuple[list[Message], dict[str, Any], ChannelSession | None]:
|
||||
"""Translate a Responses-API request body into Agent Framework constructs.
|
||||
|
||||
Returns a triple ``(messages, options, session)`` where:
|
||||
|
||||
- ``messages`` is the parsed conversation.
|
||||
- ``options`` is a ``ChatOptions``-shaped dict with the model-tunable
|
||||
fields the channel lifted off the body.
|
||||
- ``session`` is a :class:`ChannelSession` keyed by
|
||||
``previous_response_id`` when one was supplied, else ``None``.
|
||||
"""
|
||||
messages = messages_from_responses_input(body.get("input"))
|
||||
|
||||
options: dict[str, Any] = {}
|
||||
for key, value in body.items():
|
||||
if key in _RESPONSES_TRANSPORT_KEYS or value is None:
|
||||
continue
|
||||
if (mapped := _RESPONSES_OPTION_REMAP.get(key)) is not None:
|
||||
options[mapped] = value
|
||||
elif key in _RESPONSES_OPTION_PASSTHROUGH:
|
||||
options[key] = value
|
||||
# silently drop everything else (truncation, reasoning, include, ...)
|
||||
|
||||
session: ChannelSession | None = None
|
||||
if (prev := body.get("previous_response_id")) and isinstance(prev, str):
|
||||
session = ChannelSession(isolation_key=prev)
|
||||
|
||||
return messages, options, session
|
||||
|
||||
|
||||
__all__ = [
|
||||
"messages_from_responses_input",
|
||||
"parse_response_target",
|
||||
"parse_responses_identity",
|
||||
"parse_responses_request",
|
||||
]
|
||||
Reference in New Issue
Block a user