test(provider): cover OpenAI Responses adapter

astrbot/core/provider/sources/openai_responses_source.py

@@ -26,6 +26,23 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
             self.client.responses.create,
         ).parameters.keys()
 
+    @staticmethod
+    def _get_field(obj: Any, name: str, default: Any = None) -> Any:
+        if isinstance(obj, dict):
+            return obj.get(name, default)
+        return getattr(obj, name, default)
+
+    @staticmethod
+    def _arguments_to_json_string(arguments: Any) -> str:
+        if arguments is None:
+            return ""
+        if isinstance(arguments, str):
+            return arguments
+        try:
+            return json.dumps(arguments, ensure_ascii=False)
+        except TypeError:
+            return str(arguments)
+
     @staticmethod
     def _message_content_to_response_content(content: Any, role: str) -> Any:
         if isinstance(content, str) or content is None:
@@ -74,29 +91,20 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
 
     @staticmethod
     def _chat_tool_call_to_response_function_call(tool_call: Any) -> dict:
-        if isinstance(tool_call, dict):
-            function = tool_call.get("function", {})
-            call_id = tool_call.get("id") or tool_call.get("call_id") or ""
-        else:
-            function = getattr(tool_call, "function", {})
-            call_id = (
-                getattr(tool_call, "id", None)
-                or getattr(tool_call, "call_id", None)
-                or ""
-            )
-
-        if isinstance(function, dict):
-            name = function.get("name", "")
-            arguments = function.get("arguments", "")
-        else:
-            name = getattr(function, "name", "")
-            arguments = getattr(function, "arguments", "")
+        function = ProviderOpenAIResponses._get_field(tool_call, "function", {})
+        call_id = (
+            ProviderOpenAIResponses._get_field(tool_call, "id")
+            or ProviderOpenAIResponses._get_field(tool_call, "call_id")
+            or ""
+        )
+        name = ProviderOpenAIResponses._get_field(function, "name", "")
+        arguments = ProviderOpenAIResponses._get_field(function, "arguments", "")
 
         return {
             "type": "function_call",
             "call_id": call_id,
             "name": name or "",
-            "arguments": arguments or "",
+            "arguments": ProviderOpenAIResponses._arguments_to_json_string(arguments),
             "status": "completed",
         }
 
@@ -173,24 +181,16 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
             return None
 
         def _get(name: str) -> int:
-            if isinstance(usage, dict):
-                value = usage.get(name, 0)
-            else:
-                value = getattr(usage, name, 0)
+            value = ProviderOpenAIResponses._get_field(usage, name, 0)
             return value if isinstance(value, int) else 0
 
         input_tokens = _get("input_tokens")
         output_tokens = _get("output_tokens")
         cached = 0
-        details = (
-            usage.get("input_tokens_details")
-            if isinstance(usage, dict)
-            else getattr(usage, "input_tokens_details", None)
-        )
-        if isinstance(details, dict):
-            cached = details.get("cached_tokens", 0) or 0
-        elif details is not None:
-            cached = getattr(details, "cached_tokens", 0) or 0
+        details = ProviderOpenAIResponses._get_field(usage, "input_tokens_details")
+        if details is not None:
+            cached = ProviderOpenAIResponses._get_field(details, "cached_tokens", 0)
+            cached = cached or 0
         cached = cached if isinstance(cached, int) else 0
         return TokenUsage(
             input_other=max(input_tokens - cached, 0),
@@ -200,105 +200,77 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
 
     @staticmethod
     def _extract_response_output_text(response: Any) -> str:
-        output_text = getattr(response, "output_text", None)
+        output_text = ProviderOpenAIResponses._get_field(response, "output_text")
         if isinstance(output_text, str):
             return output_text.strip()
-        if isinstance(response, dict) and isinstance(response.get("output_text"), str):
-            return response["output_text"].strip()
 
-        output = (
-            response.get("output", [])
-            if isinstance(response, dict)
-            else getattr(response, "output", [])
-        )
+        output = ProviderOpenAIResponses._get_field(response, "output", [])
         parts: list[str] = []
         if isinstance(output, list):
             for item in output:
-                content = (
-                    item.get("content", [])
-                    if isinstance(item, dict)
-                    else getattr(item, "content", [])
-                )
+                content = ProviderOpenAIResponses._get_field(item, "content", [])
                 if not isinstance(content, list):
                     continue
                 for part in content:
-                    part_type = (
-                        part.get("type")
-                        if isinstance(part, dict)
-                        else getattr(part, "type", None)
-                    )
+                    part_type = ProviderOpenAIResponses._get_field(part, "type")
                     if part_type not in {"output_text", "text"}:
                         continue
-                    text = (
-                        part.get("text")
-                        if isinstance(part, dict)
-                        else getattr(part, "text", None)
-                    )
+                    text = ProviderOpenAIResponses._get_field(part, "text")
                     if isinstance(text, str):
                         parts.append(text)
         return "".join(parts).strip()
 
     @staticmethod
     def _iter_response_output_items(response: Any) -> list[Any]:
-        if isinstance(response, dict):
-            output = response.get("output", [])
-        else:
-            output = getattr(response, "output", [])
+        output = ProviderOpenAIResponses._get_field(response, "output", [])
         return output if isinstance(output, list) else []
 
+    @classmethod
+    def _iter_function_calls(cls, response: Any) -> list[dict[str, Any]]:
+        calls: list[dict[str, Any]] = []
+        for item in cls._iter_response_output_items(response):
+            if cls._get_field(item, "type") != "function_call":
+                continue
+            calls.append(
+                {
+                    "name": cls._get_field(item, "name"),
+                    "arguments": cls._get_field(item, "arguments"),
+                    "call_id": cls._get_field(item, "call_id"),
+                }
+            )
+        return calls
+
+    @staticmethod
+    def _parse_function_call_arguments(arguments: Any) -> dict:
+        if isinstance(arguments, str):
+            try:
+                parsed_args = json.loads(arguments)
+            except json.JSONDecodeError:
+                return {}
+            return parsed_args if isinstance(parsed_args, dict) else {}
+        if isinstance(arguments, dict):
+            return arguments
+        return {}
+
     async def _parse_responses_completion(
         self,
         response: Any,
         tools: ToolSet | None,
     ) -> LLMResponse:
         llm_response = LLMResponse("assistant")
-        response_id = (
-            response.get("id")
-            if isinstance(response, dict)
-            else getattr(response, "id", None)
-        )
+        response_id = self._get_field(response, "id")
 
         if tools is not None:
             args_ls: list[dict] = []
             func_name_ls: list[str] = []
             tool_call_ids: list[str] = []
-            for item in self._iter_response_output_items(response):
-                item_type = (
-                    item.get("type")
-                    if isinstance(item, dict)
-                    else getattr(item, "type", None)
-                )
-                if item_type != "function_call":
-                    continue
-                name = (
-                    item.get("name")
-                    if isinstance(item, dict)
-                    else getattr(item, "name", None)
-                )
-                arguments = (
-                    item.get("arguments")
-                    if isinstance(item, dict)
-                    else getattr(item, "arguments", None)
-                )
-                call_id = (
-                    item.get("call_id")
-                    if isinstance(item, dict)
-                    else getattr(item, "call_id", None)
-                )
+            for call in self._iter_function_calls(response):
+                name = call["name"]
                 if not name:
                     continue
-                if isinstance(arguments, str):
-                    try:
-                        parsed_args = json.loads(arguments)
-                    except json.JSONDecodeError:
-                        parsed_args = {}
-                elif isinstance(arguments, dict):
-                    parsed_args = arguments
-                else:
-                    parsed_args = {}
-                args_ls.append(parsed_args)
+                args_ls.append(self._parse_function_call_arguments(call["arguments"]))
                 func_name_ls.append(name)
-                tool_call_ids.append(call_id or response_id or "")
+                tool_call_ids.append(call["call_id"] or response_id or "")
             if args_ls:
                 llm_response.role = "tool"
                 llm_response.tools_call_args = args_ls
@@ -310,11 +282,7 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
             llm_response.result_chain = MessageChain().message(completion_text)
         llm_response.raw_completion = response
         llm_response.id = response_id
-        usage = (
-            response.get("usage")
-            if isinstance(response, dict)
-            else getattr(response, "usage", None)
-        )
+        usage = self._get_field(response, "usage")
         llm_response.usage = self._response_usage_to_token_usage(usage)
         return llm_response
 
@@ -331,13 +299,11 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
         self._apply_provider_specific_extra_body_overrides(extra_body)
         return request_payload, extra_body
 
-    async def _query(
+    def _build_responses_request(
         self,
         payloads: dict,
         tools: ToolSet | None,
-        *,
-        request_max_retries: int | None = None,
-    ) -> LLMResponse:
+    ) -> tuple[dict, dict]:
         self._sanitize_assistant_messages(payloads)
         response_payload = self._chat_payload_to_responses_payload(payloads)
         response_tools = self._responses_function_tools(tools)
@@ -349,8 +315,16 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
                 )
         else:
             response_payload.pop("tool_choice", None)
+        return self._split_responses_extra_body(response_payload)
 
-        request_payload, extra_body = self._split_responses_extra_body(response_payload)
+    async def _query(
+        self,
+        payloads: dict,
+        tools: ToolSet | None,
+        *,
+        request_max_retries: int | None = None,
+    ) -> LLMResponse:
+        request_payload, extra_body = self._build_responses_request(payloads, tools)
         response = await retry_provider_request(
             "OpenAI",
             lambda: self.client.responses.create(
@@ -364,9 +338,75 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
 
     @staticmethod
     def _event_value(event: Any, name: str, default: Any = None) -> Any:
-        if isinstance(event, dict):
-            return event.get(name, default)
-        return getattr(event, name, default)
+        return ProviderOpenAIResponses._get_field(event, name, default)
+
+    @classmethod
+    def _stream_function_call_key(
+        cls,
+        event: Any,
+        function_calls: dict[str, dict[str, Any]],
+    ) -> str:
+        item = cls._event_value(event, "item")
+        for value in (
+            cls._event_value(event, "output_index"),
+            cls._event_value(event, "item_id"),
+            cls._get_field(item, "id"),
+            cls._get_field(item, "call_id"),
+        ):
+            if value is not None:
+                return str(value)
+        return str(len(function_calls))
+
+    @classmethod
+    def _merge_stream_function_call_event(
+        cls,
+        event: Any,
+        function_calls: dict[str, dict[str, Any]],
+    ) -> None:
+        event_type = cls._event_value(event, "type", "")
+        item = cls._event_value(event, "item")
+        call_key = cls._stream_function_call_key(event, function_calls)
+
+        if event_type in {"response.output_item.added", "response.output_item.done"}:
+            if cls._get_field(item, "type") != "function_call":
+                return
+            call = function_calls.setdefault(call_key, {})
+            call["name"] = cls._get_field(item, "name", call.get("name"))
+            call["call_id"] = cls._get_field(item, "call_id", call.get("call_id"))
+            arguments = cls._get_field(item, "arguments")
+            if arguments is not None:
+                call["arguments"] = arguments
+            return
+
+        if event_type == "response.function_call_arguments.delta":
+            delta = cls._event_value(event, "delta", "")
+            if delta:
+                call = function_calls.setdefault(call_key, {})
+                call["arguments"] = f"{call.get('arguments', '')}{delta}"
+            return
+
+        if event_type == "response.function_call_arguments.done":
+            arguments = cls._event_value(event, "arguments", "")
+            function_calls.setdefault(call_key, {})["arguments"] = arguments
+
+    async def _stream_function_calls_to_response(
+        self,
+        function_calls: dict[str, dict[str, Any]],
+        tools: ToolSet | None,
+    ) -> LLMResponse:
+        output = []
+        for call in function_calls.values():
+            if not call.get("name"):
+                continue
+            output.append(
+                {
+                    "type": "function_call",
+                    "name": call.get("name", ""),
+                    "call_id": call.get("call_id", ""),
+                    "arguments": call.get("arguments", ""),
+                }
+            )
+        return await self._parse_responses_completion({"output": output}, tools)
 
     async def _query_stream(
         self,
@@ -375,19 +415,7 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
         *,
         request_max_retries: int | None = None,
     ) -> AsyncGenerator[LLMResponse, None]:
-        self._sanitize_assistant_messages(payloads)
-        response_payload = self._chat_payload_to_responses_payload(payloads)
-        response_tools = self._responses_function_tools(tools)
-        if response_tools:
-            response_payload["tools"] = response_tools
-            if tools and not tools.empty():
-                response_payload["tool_choice"] = response_payload.get(
-                    "tool_choice", "auto"
-                )
-        else:
-            response_payload.pop("tool_choice", None)
-
-        request_payload, extra_body = self._split_responses_extra_body(response_payload)
+        request_payload, extra_body = self._build_responses_request(payloads, tools)
         stream = await retry_provider_request(
             "OpenAI",
             lambda: self.client.responses.create(
@@ -400,6 +428,7 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
 
         output_text = ""
         final_response = None
+        function_calls: dict[str, dict[str, Any]] = {}
         async for event in stream:
             event_type = self._event_value(event, "type", "")
             if event_type == "response.output_text.delta":
@@ -418,11 +447,18 @@ class ProviderOpenAIResponses(ProviderOpenAIOfficial):
                     output_text = str(text)
             elif event_type == "response.completed":
                 final_response = self._event_value(event, "response")
+            else:
+                self._merge_stream_function_call_event(event, function_calls)
 
         if final_response is not None:
             llm_response = await self._parse_responses_completion(final_response, tools)
             if not llm_response.completion_text and output_text:
                 llm_response.result_chain = MessageChain().message(output_text)
+        elif function_calls:
+            llm_response = await self._stream_function_calls_to_response(
+                function_calls,
+                tools,
+            )
         else:
             llm_response = LLMResponse(
                 "assistant",

tests/test_openai_responses_source.py

@@ -0,0 +1,213 @@
+from types import SimpleNamespace
+
+import pytest
+
+from astrbot.core.agent.tool import FunctionTool, ToolSet
+from astrbot.core.provider.sources.openai_responses_source import (
+    ProviderOpenAIResponses,
+)
+
+
+class _Responses:
+    async def create(self, **kwargs):
+        return SimpleNamespace(output_text="ok", output=[], usage=None)
+
+
+class _Client:
+    def __init__(self) -> None:
+        self.responses = _Responses()
+
+
+def _make_provider() -> ProviderOpenAIResponses:
+    provider = ProviderOpenAIResponses.__new__(ProviderOpenAIResponses)
+    provider.client = _Client()
+    provider.default_params = {
+        "model",
+        "input",
+        "tools",
+        "tool_choice",
+        "stream",
+        "extra_body",
+    }
+    provider.provider_config = {"custom_extra_body": {"metadata": {"test": True}}}
+    provider._apply_provider_specific_extra_body_overrides = lambda extra_body: None
+    return provider
+
+
+def _make_tool_set() -> ToolSet:
+    return ToolSet(
+        tools=[
+            FunctionTool(
+                name="lookup_weather",
+                description="Look up weather",
+                parameters={
+                    "type": "object",
+                    "properties": {"city": {"type": "string"}},
+                    "required": ["city"],
+                },
+            )
+        ]
+    )
+
+
+def test_chat_payload_to_responses_payload_converts_messages_and_tool_calls():
+    payload = {
+        "model": "gpt-4.1",
+        "messages": [
+            {"role": "system", "content": "Be brief."},
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "weather"},
+                    {"type": "image_url", "image_url": {"url": "data:image/png,abc"}},
+                ],
+            },
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "function": {
+                            "name": "lookup_weather",
+                            "arguments": {"city": "Shanghai"},
+                        },
+                    }
+                ],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "sunny"},
+        ],
+    }
+
+    converted = ProviderOpenAIResponses._chat_payload_to_responses_payload(payload)
+
+    assert "messages" not in converted
+    assert converted["model"] == "gpt-4.1"
+    assert converted["input"] == [
+        {"role": "system", "content": "Be brief."},
+        {
+            "role": "user",
+            "content": [
+                {"type": "input_text", "text": "weather"},
+                {"type": "input_image", "image_url": "data:image/png,abc"},
+            ],
+        },
+        {
+            "type": "function_call",
+            "call_id": "call_1",
+            "name": "lookup_weather",
+            "arguments": '{"city": "Shanghai"}',
+            "status": "completed",
+        },
+        {"type": "function_call_output", "call_id": "call_1", "output": "sunny"},
+    ]
+
+
+def test_build_responses_request_shares_tool_and_extra_body_handling():
+    provider = _make_provider()
+    payload = {
+        "model": "gpt-4.1",
+        "messages": [{"role": "user", "content": "hello"}],
+        "unknown_param": "kept-in-extra-body",
+    }
+
+    request_payload, extra_body = provider._build_responses_request(
+        payload,
+        _make_tool_set(),
+    )
+
+    assert request_payload["input"] == [{"role": "user", "content": "hello"}]
+    assert request_payload["tool_choice"] == "auto"
+    assert request_payload["tools"] == [
+        {
+            "type": "function",
+            "name": "lookup_weather",
+            "description": "Look up weather",
+            "parameters": {
+                "type": "object",
+                "properties": {"city": {"type": "string"}},
+                "required": ["city"],
+            },
+            "strict": False,
+        }
+    ]
+    assert extra_body == {
+        "metadata": {"test": True},
+        "unknown_param": "kept-in-extra-body",
+    }
+
+
+@pytest.mark.asyncio
+async def test_parse_responses_completion_extracts_function_call_and_usage():
+    provider = _make_provider()
+    response = SimpleNamespace(
+        id="resp_1",
+        output=[
+            SimpleNamespace(
+                type="function_call",
+                name="lookup_weather",
+                call_id="call_1",
+                arguments='{"city":"Guangzhou"}',
+            )
+        ],
+        usage=SimpleNamespace(
+            input_tokens=10,
+            output_tokens=3,
+            input_tokens_details=SimpleNamespace(cached_tokens=4),
+        ),
+    )
+
+    result = await provider._parse_responses_completion(response, _make_tool_set())
+
+    assert result.role == "tool"
+    assert result.tools_call_name == ["lookup_weather"]
+    assert result.tools_call_ids == ["call_1"]
+    assert result.tools_call_args == [{"city": "Guangzhou"}]
+    assert result.usage.input_other == 6
+    assert result.usage.input_cached == 4
+    assert result.usage.output == 3
+
+
+@pytest.mark.asyncio
+async def test_stream_function_call_events_are_converted_to_tool_response():
+    provider = _make_provider()
+    function_calls: dict[str, dict] = {}
+
+    ProviderOpenAIResponses._merge_stream_function_call_event(
+        {
+            "type": "response.output_item.added",
+            "output_index": 0,
+            "item": {
+                "type": "function_call",
+                "name": "lookup_weather",
+                "call_id": "call_1",
+            },
+        },
+        function_calls,
+    )
+    ProviderOpenAIResponses._merge_stream_function_call_event(
+        {
+            "type": "response.function_call_arguments.delta",
+            "output_index": 0,
+            "delta": '{"city"',
+        },
+        function_calls,
+    )
+    ProviderOpenAIResponses._merge_stream_function_call_event(
+        {
+            "type": "response.function_call_arguments.delta",
+            "output_index": 0,
+            "delta": ':"Shanghai"}',
+        },
+        function_calls,
+    )
+
+    result = await provider._stream_function_calls_to_response(
+        function_calls,
+        _make_tool_set(),
+    )
+
+    assert result.role == "tool"
+    assert result.tools_call_name == ["lookup_weather"]
+    assert result.tools_call_ids == ["call_1"]
+    assert result.tools_call_args == [{"city": "Shanghai"}]