feat(provider): add OpenAI Responses provider

[DolphinZZZZZ]

Summary

• add a separate openai_responses provider adapter that uses client.responses.create
• add an OpenAI Response provider template with the same configurable fields as OpenAI Compatible
• keep the existing OpenAI Compatible chat-completions provider unchanged

Scope

This does not rewrite or normalize api_base; the configured base URL is passed through to the OpenAI SDK as before. Existing OpenAI-compatible providers continue to use chat completions.

Test

• python -m py_compile astrbot/core/provider/sources/openai_responses_source.py astrbot/core/provider/manager.py astrbot/core/config/default.py
• git diff --check

Summary by Sourcery

Add a new OpenAI Responses-based provider adapter and wire it into the provider manager and default configuration.

New Features:

• Introduce an OpenAI Responses provider adapter that uses the Responses API while remaining compatible with the existing provider abstraction.
• Add an OpenAI Response provider entry to the default configuration that mirrors the OpenAI-compatible chat completions settings.

Enhancements:

• Extend provider manager dynamic import logic to load the new OpenAI Responses adapter type.

[vercel]

@DolphinZZZZZ is attempting to deploy a commit to the soulter's projects Team on Vercel.

A member of the Team first needs to authorize it.

[gemini-code-assist]

Code Review

This pull request introduces a new openai_responses provider adapter to support the OpenAI Responses API, including its configuration and dynamic loading. Feedback on these changes suggests adding unit tests for the new adapter to ensure correctness and prevent regressions, as well as serializing function call arguments to a JSON string to avoid API validation errors.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The new openai_responses provider adapter introduces significant new functionality, including complex payload conversion and response parsing. Please accompany this with corresponding unit tests to ensure correctness and prevent regressions.

References

1. New functionality, such as handling attachments, should be accompanied by corresponding unit tests.

[gemini-code-assist]

The OpenAI Responses API expects arguments for a function call to be a JSON string. If arguments is passed as a dictionary or other non-string object, it should be serialized to a JSON string to prevent API validation errors.

        if isinstance(function, dict):
            name = function.get("name", "")
            arguments = function.get("arguments", "")
        else:
            name = getattr(function, "name", "")
            arguments = getattr(function, "arguments", "")

        if not isinstance(arguments, str) and arguments is not None:
            try:
                arguments = json.dumps(arguments, ensure_ascii=False)
            except Exception:
                arguments = ""

        return {
            "type": "function_call",
            "call_id": call_id,
            "name": name or "",
            "arguments": arguments or "",
            "status": "completed",
        }

[sourcery-ai]

Hey - I've found 1 issue, and left some high level feedback:

• The streaming path in _query_stream only handles response.output_text.* and response.completed events, so any streamed tool/function call events from the Responses API are currently ignored; consider handling those to keep tool invocation behavior consistent with the non-streaming _query path.
• Both _query and _query_stream duplicate the same payload preparation logic (sanitizing assistant messages, mapping chat payload to responses payload, tools conversion, and extra_body splitting); consider extracting this into a shared helper to reduce divergence risk between streaming and non-streaming behavior.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The streaming path in `_query_stream` only handles `response.output_text.*` and `response.completed` events, so any streamed tool/function call events from the Responses API are currently ignored; consider handling those to keep tool invocation behavior consistent with the non-streaming `_query` path.
- Both `_query` and `_query_stream` duplicate the same payload preparation logic (sanitizing assistant messages, mapping chat payload to responses payload, tools conversion, and extra_body splitting); consider extracting this into a shared helper to reduce divergence risk between streaming and non-streaming behavior.

## Individual Comments

### Comment 1
<location path="astrbot/core/provider/sources/openai_responses_source.py" line_range="20" />
<code_context>
+    "openai_responses",
+    "OpenAI Responses API 提供商适配器",
+)
+class ProviderOpenAIResponses(ProviderOpenAIOfficial):
+    """OpenAI-compatible provider that calls the Responses API."""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring common patterns (field access, request setup, and tool-call parsing) into shared helpers to reduce duplication and clarify the control flow.

You can reduce the complexity meaningfully without changing behavior by introducing a couple of small normalizing helpers and sharing the request-building logic.

### 1. Unify dict/object access

You have the same `dict` vs object branching in multiple places (`_response_usage_to_token_usage`, `_extract_response_output_text`, `_iter_response_output_items`, `_parse_responses_completion`, `_event_value`, `_chat_tool_call_to_response_function_call`).

Introduce a tiny helper and use it everywhere instead of re-implementing the pattern:

```python
@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)
```

Then, for example:

```python
@staticmethod
def _response_usage_to_token_usage(usage: Any) -> TokenUsage | None:
    if not usage:
        return None

    def _get_int(name: str) -> int:
        value = ProviderOpenAIResponses._get_field(usage, name, 0)
        return value if isinstance(value, int) else 0

    input_tokens = _get_int("input_tokens")
    output_tokens = _get_int("output_tokens")

    details = ProviderOpenAIResponses._get_field(usage, "input_tokens_details")
    cached = ProviderOpenAIResponses._get_field(details, "cached_tokens", 0) if details else 0
    cached = cached if isinstance(cached, int) else 0

    return TokenUsage(
        input_other=max(input_tokens - cached, 0),
        input_cached=cached,
        output=output_tokens,
    )
```

And in `_extract_response_output_text`:

```python
@staticmethod
def _extract_response_output_text(response: Any) -> str:
    output_text = ProviderOpenAIResponses._get_field(response, "output_text")
    if isinstance(output_text, str):
        return output_text.strip()

    output = ProviderOpenAIResponses._get_field(response, "output", [])
    parts: list[str] = []
    if isinstance(output, list):
        for item in output:
            content = ProviderOpenAIResponses._get_field(item, "content", [])
            if not isinstance(content, list):
                continue
            for part in content:
                part_type = ProviderOpenAIResponses._get_field(part, "type")
                if part_type not in {"output_text", "text"}:
                    continue
                text = ProviderOpenAIResponses._get_field(part, "text")
                if isinstance(text, str):
                    parts.append(text)
    return "".join(parts).strip()
```

You can similarly simplify `_iter_response_output_items`, `_event_value`, `_chat_tool_call_to_response_function_call`, and the tool call parsing loop in `_parse_responses_completion`, which removes quite a bit of repeated branching.

### 2. Share request-building between `_query` and `_query_stream`

Right now `_query` and `_query_stream` share almost all of the setup code. You can factor that into a small helper that returns `request_payload` and `extra_body`, leaving only the stream vs non-stream behavior different.

```python
def _build_responses_request(
    self,
    payloads: dict,
    tools: ToolSet | None,
) -> tuple[dict, dict]:
    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)

    return self._split_responses_extra_body(response_payload)
```

Then `_query` / `_query_stream` reduce to:

```python
async def _query(...):
    request_payload, extra_body = self._build_responses_request(payloads, tools)
    response = await retry_provider_request(
        "OpenAI",
        lambda: self.client.responses.create(
            **request_payload,
            stream=False,
            extra_body=extra_body,
        ),
        max_attempts=request_max_retries,
    )
    return await self._parse_responses_completion(response, tools)
```

```python
async def _query_stream(...):
    request_payload, extra_body = self._build_responses_request(payloads, tools)
    stream = await retry_provider_request(
        "OpenAI",
        lambda: self.client.responses.create(
            **request_payload,
            stream=True,
            extra_body=extra_body,
        ),
        max_attempts=request_max_retries,
    )
    ...
```

This keeps all current behavior but localizes the complex setup logic in one place.

### 3. Normalize tool call extraction

`_chat_tool_call_to_response_function_call` and the loop inside `_parse_responses_completion` implement very similar extraction logic for `name`, `arguments`, and `call_id`. You can centralize that into a single iterator returning normalized objects:

```python
@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):
        item_type = cls._get_field(item, "type")
        if item_type != "function_call":
            continue
        name = cls._get_field(item, "name")
        arguments = cls._get_field(item, "arguments")
        call_id = cls._get_field(item, "call_id")
        calls.append({"name": name, "arguments": arguments, "call_id": call_id})
    return calls
```

Then `_parse_responses_completion` simplifies to:

```python
if tools is not None:
    args_ls: list[dict] = []
    func_name_ls: list[str] = []
    tool_call_ids: list[str] = []

    for call in self._iter_function_calls(response):
        name = call["name"]
        if not name:
            continue
        raw_args = call["arguments"]
        if isinstance(raw_args, str):
            try:
                parsed_args = json.loads(raw_args)
            except json.JSONDecodeError:
                parsed_args = {}
        elif isinstance(raw_args, dict):
            parsed_args = raw_args
        else:
            parsed_args = {}

        args_ls.append(parsed_args)
        func_name_ls.append(name)
        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
        llm_response.tools_call_name = func_name_ls
        llm_response.tools_call_ids = tool_call_ids
```

This keeps the semantics identical but removes duplicated extraction logic and makes tool handling more coherent.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}