feat: add stateless Responses API provider - #9515
Merged
Merged
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Deploying with
|
| Status | Name | Latest Commit | Preview URL | Updated (UTC) |
|---|---|---|---|---|
| ✅ Deployment successful! View logs |
astrbot-docs | 9c75719 | Commit Preview URL Branch Preview URL |
Aug 02 2026, 03:16 PM |
…ring sanitization' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Soulter
marked this pull request as ready for review
August 2, 2026 15:09
dosubot
Bot
added
size:XL
📄 Knowledge reviewDosu skipped reviewing this PR because your organization has used its
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- The logic that normalizes payloads and extra_body (tools flattening, max_tokens/reasoning_effort mapping, and stateless store/conversation/previous_response_id handling) is duplicated in _query and _query_stream; consider extracting this into a shared helper to reduce maintenance overhead and keep behavior in sync.
- _convert_chat_messages_to_response_input silently drops messages with unsupported roles and non-dict content parts; if this is not intentional, you may want more explicit handling or logging to surface unexpected history shapes instead of silently discarding them.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The logic that normalizes payloads and extra_body (tools flattening, max_tokens/reasoning_effort mapping, and stateless store/conversation/previous_response_id handling) is duplicated in _query and _query_stream; consider extracting this into a shared helper to reduce maintenance overhead and keep behavior in sync.
- _convert_chat_messages_to_response_input silently drops messages with unsupported roles and non-dict content parts; if this is not intentional, you may want more explicit handling or logging to surface unexpected history shapes instead of silently discarding them.
## Individual Comments
### Comment 1
<location path="astrbot/core/provider/sources/openai_responses_source.py" line_range="559-568" />
<code_context>
+ if item_type == "function_call" and tools is not None:
</code_context>
<issue_to_address>
**suggestion:** Responses API function call outputs and missing tool names/call IDs are not guarded against, which may cause subtle tool-calling issues.
In `_parse_response`, tool handling is limited to `item_type == "function_call"` and assumes non-empty `name` and `call_id`. The Responses API may also emit `function_call_output` or entries with missing/invalid `call_id`. Please either ignore or log unexpected tool item types (e.g., `function_call_output`) and add checks for missing/empty `name` and `call_id` so invalid tool calls are not added to `LLMResponse`.
Suggested implementation:
```python
if item_type in ("function_call", "function_call_output") and tools is not None:
# Ignore unexpected tool item types (e.g., function_call_output) but log them for debugging.
if item_type != "function_call":
logger.warning("Ignoring unexpected tool item type in response: %s", item_type)
continue
# Validate that we have a usable tool name and call_id; otherwise, skip this tool call.
name = self._field(item, "name")
call_id = self._field(item, "call_id")
if not name or not call_id:
logger.warning(
"Skipping tool call with missing or empty name/call_id: name=%r, call_id=%r",
name,
call_id,
)
continue
arguments = self._field(item, "arguments", "{}")
if isinstance(arguments, str):
try:
parsed_arguments = json.loads(arguments)
except json.JSONDecodeError as exc:
logger.error("Failed to parse function arguments: %s", exc)
parsed_arguments = {}
else:
parsed_arguments = arguments
if parsed_arguments is None:
```
1. If later in this function you create tool call entries for `LLMResponse` by re-reading `name` or `call_id` from `item`, you do not need to change that logic; the new validation already ensures only valid tool items reach that part of the code.
2. If you have or add metrics/structured logging, you may want to replace the `logger.warning` calls with your standard logging/telemetry helpers to ensure these invalid/ignored tool events are captured consistently.
</issue_to_address>
### Comment 2
<location path="tests/test_openai_responses_source.py" line_range="50-61" />
<code_context>
+ return Response.model_validate(payload)
+
+
+def test_responses_provider_templates_are_independent_and_stateless():
+ templates = CONFIG_METADATA_2["provider_group"]["metadata"]["provider"][
+ "config_template"
+ ]
+
+ assert templates["OpenAI Responses"]["type"] == "openai_responses"
+ assert templates["OpenAI Responses"]["api_base"] == "https://api.openai.com/v1"
+ assert templates["DeepSeek Responses"]["type"] == "openai_responses"
+ assert templates["DeepSeek Responses"]["api_base"] == "https://api.deepseek.com/v1"
+ assert templates["xAI"]["type"] == "openai_responses"
+ assert templates["xAI"]["api_base"] == "https://api.x.ai/v1"
+ assert "xai_native_search" not in templates["xAI"]
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add explicit tests for the xAI chat-completions template and conditions to guard against regressions.
This test covers the new Responses-based xAI template, but `CONFIG_METADATA_2` still defines the legacy `xai_chat_completion` template and `xai_native_search` flag. Please add assertions that the legacy "xAI Chat Completions" template is still present, has `type == "xai_chat_completion"`, and that `xai_native_search` is scoped only to that template. This will better protect against regressions in the legacy search behavior during the Responses migration.
```suggestion
def test_responses_provider_templates_are_independent_and_stateless():
templates = CONFIG_METADATA_2["provider_group"]["metadata"]["provider"][
"config_template"
]
# Responses-based templates
assert templates["OpenAI Responses"]["type"] == "openai_responses"
assert templates["OpenAI Responses"]["api_base"] == "https://api.openai.com/v1"
assert "xai_native_search" not in templates["OpenAI Responses"]
assert templates["DeepSeek Responses"]["type"] == "openai_responses"
assert templates["DeepSeek Responses"]["api_base"] == "https://api.deepseek.com/v1"
assert "xai_native_search" not in templates["DeepSeek Responses"]
assert templates["xAI"]["type"] == "openai_responses"
assert templates["xAI"]["api_base"] == "https://api.x.ai/v1"
assert "xai_native_search" not in templates["xAI"]
# Legacy xAI chat-completions template must remain present and correctly typed
assert "xAI Chat Completions" in templates
assert templates["xAI Chat Completions"]["type"] == "xai_chat_completion"
# The legacy xai_native_search flag is scoped only to the chat-completions template
assert "xai_native_search" in templates["xAI Chat Completions"]
```
</issue_to_address>
### Comment 3
<location path="astrbot/core/provider/sources/openai_responses_source.py" line_range="59" />
<code_context>
+ return value.get(name, default)
+ return getattr(value, name, default)
+
+ def _convert_chat_messages_to_response_input(
+ self,
+ messages: list[dict],
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting small helper methods for message conversion, payload shaping, response parsing, and retry payload reconstruction to keep this adapter’s core methods linear and easier to maintain.
You can keep all current behavior while reducing complexity with a few small, focused helpers.
### 1. Reduce branching in `_convert_chat_messages_to_response_input`
Split out user/assistant content conversion and reasoning extraction; keep the main method as orchestration:
```python
def _is_deepseek(self) -> bool:
host = (self.client.base_url.host or "").rstrip(".").lower()
return (
self.provider_config.get("provider") == "deepseek"
or host == "api.deepseek.com"
)
def _extract_reasoning_items_from_content(
self, content: list[dict], is_deepseek: bool
) -> list[dict]:
reasoning_items: list[dict] = []
for part in content:
if not isinstance(part, dict):
continue
if part.get("type") != "think":
continue
serialized_state = part.get("encrypted")
restored_items: list[dict] = []
if isinstance(serialized_state, str):
try:
state = json.loads(serialized_state)
except json.JSONDecodeError:
state = None
if (
isinstance(state, dict)
and state.get("type") == self._REASONING_STATE_TYPE
and isinstance(state.get("items"), list)
):
restored_items = [
item for item in state["items"] if isinstance(item, dict)
]
if restored_items:
reasoning_items.extend(restored_items)
elif is_deepseek and part.get("think"):
reasoning_items.append(
{
"type": "reasoning",
"content": [
{
"type": "reasoning_text",
"text": str(part["think"]),
}
],
"summary": [],
}
)
return reasoning_items
```
Then simplify `_convert_chat_messages_to_response_input` by delegating:
```python
def _convert_chat_messages_to_response_input(
self, messages: list[dict],
) -> list[dict]:
response_input: list[dict] = []
is_deepseek = self._is_deepseek()
for message in messages:
# tool/tool_call_output handling unchanged ...
role = message.get("role")
if role not in {"user", "assistant", "system", "developer"}:
continue
content = message.get("content")
reasoning_items: list[dict] = []
converted_content: str | list[dict] | None = None
if isinstance(content, list):
reasoning_items = self._extract_reasoning_items_from_content(
content, is_deepseek
)
# small helper can convert text/image/audio parts:
converted_content = self._convert_user_content_parts(content, role)
elif isinstance(content, str):
converted_content = content
elif content is not None:
converted_content = str(content)
response_input.extend(reasoning_items)
if converted_content not in (None, "", []):
response_input.append(
{
"type": "message",
"role": role,
"content": converted_content,
}
)
if role == "assistant":
self._append_tool_calls(response_input, message)
return response_input
```
You already have the logic for tool calls; pulling it into `_append_tool_calls` keeps this method linear and easier to maintain.
### 2. Deduplicate `_query` / `_query_stream` payload shaping
The tools/extra_body handling is almost identical. A shared helper keeps both methods focused on transport:
```python
def _prepare_responses_request(
self,
payloads: dict[str, Any],
tools: ToolSet | None,
) -> tuple[dict[str, Any], dict[str, Any]]:
payloads = dict(payloads) # shallow copy to avoid mutating caller
if tools:
response_tools = []
for tool in tools.openai_schema():
function = tool.get("function", {})
response_tools.append({"type": "function", **function})
if response_tools:
payloads["tools"] = response_tools
payloads["tool_choice"] = payloads.get("tool_choice", "auto")
extra_body: dict[str, Any] = {}
custom_extra_body = self.provider_config.get("custom_extra_body", {})
if isinstance(custom_extra_body, dict):
extra_body.update(custom_extra_body)
for key in list(payloads):
if key not in self.default_params:
extra_body[key] = payloads.pop(key)
max_tokens = extra_body.pop("max_tokens", None)
if max_tokens is not None and "max_output_tokens" not in extra_body:
extra_body["max_output_tokens"] = max_tokens
reasoning_effort = extra_body.pop("reasoning_effort", None)
if reasoning_effort is not None and "reasoning" not in extra_body:
extra_body["reasoning"] = {"effort": reasoning_effort}
for k in ("previous_response_id", "conversation", "store"):
extra_body.pop(k, None)
payloads.pop(k, None)
payloads["store"] = False
return payloads, extra_body
```
Then `_query` and `_query_stream` can be reduced to:
```python
async def _query(self, payloads: dict, tools: ToolSet | None, *, request_max_retries: int | None = None) -> LLMResponse:
payloads, extra_body = self._prepare_responses_request(payloads, tools)
response = await retry_provider_request(
"OpenAI Responses",
lambda: self.client.responses.create(
**payloads,
stream=False,
extra_body=extra_body,
),
max_attempts=request_max_retries,
)
# type check + parse as today...
async def _query_stream(self, payloads: dict, tools: ToolSet | None, *, request_max_retries: int | None = None) -> AsyncGenerator[LLMResponse, None]:
payloads, extra_body = self._prepare_responses_request(payloads, tools)
stream = await retry_provider_request(
"OpenAI Responses",
lambda: self.client.responses.create(
**payloads,
stream=True,
extra_body=extra_body,
),
max_attempts=request_max_retries,
)
# streaming event loop as today...
```
This removes duplication and makes future changes to payload shaping much safer.
### 3. Isolate parsing responsibilities in `_parse_response`
You can keep `_parse_response` as the orchestrator and move the inner loops into helpers without altering behavior:
```python
def _extract_output_and_reasoning(
self, response: Response,
) -> tuple[list[str], list[str], list[dict]]:
text_parts: list[str] = []
reasoning_parts: list[str] = []
serialized_reasoning_items: list[dict] = []
for item in self._field(response, "output", []) or []:
item_type = self._field(item, "type")
if item_type == "message":
for content in self._field(item, "content", []) or []:
content_type = self._field(content, "type")
if content_type == "output_text":
text_parts.append(str(self._field(content, "text", "")))
elif content_type == "refusal":
text_parts.append(str(self._field(content, "refusal", "")))
elif item_type == "reasoning":
# existing serialization + summary/content handling moved here
# ...
pass
return text_parts, reasoning_parts, serialized_reasoning_items
```
Then `_parse_response` can do:
```python
text_parts, reasoning_parts, serialized_items = self._extract_output_and_reasoning(response)
completion_text = "".join(text_parts)
if completion_text:
llm_response.result_chain = MessageChain().message(completion_text)
if reasoning_parts:
llm_response.reasoning_content = "\n".join(reasoning_parts)
if serialized_items:
llm_response.reasoning_signature = json.dumps(
{"type": self._REASONING_STATE_TYPE, "items": serialized_items},
ensure_ascii=False,
)
```
Similarly, tool call extraction and usage can be moved into `_extract_tool_calls` and `_extract_usage`, shortening `_parse_response` and keeping each concern localized.
### 4. Simplify `_handle_api_error` payload reconstruction
To avoid duplicating the conversion logic and the “messages vs input” dance, introduce a helper that knows how to rebuild the Responses payload from the chat-format context:
```python
def _rebuild_responses_payload_from_context(
self, payloads: dict, context_query: list[dict],
) -> dict:
retry_payloads = dict(payloads)
retry_payloads.pop("messages", None)
retry_payloads["input"] = self._convert_chat_messages_to_response_input(
context_query
)
retry_payloads["store"] = False
return retry_payloads
```
Then `_handle_api_error` becomes:
```python
async def _handle_api_error(...):
compatibility_payloads = dict(payloads)
compatibility_payloads["messages"] = context_query
result = await super()._handle_api_error(
error,
compatibility_payloads,
context_query,
func_tool,
chosen_key,
available_api_keys,
retry_cnt,
max_retries,
image_fallback_used=image_fallback_used,
)
(
success,
chosen_key,
available_api_keys,
retry_payloads,
context_query,
func_tool,
image_fallback_used,
) = result
retry_payloads = self._rebuild_responses_payload_from_context(
retry_payloads, context_query
)
return (
success,
chosen_key,
available_api_keys,
retry_payloads,
context_query,
func_tool,
image_fallback_used,
)
```
This keeps all current behavior but makes the retry/error path easier to reason about and less coupled to the internal structure of the payload dict.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+559
to
+568
| if item_type == "function_call" and tools is not None: | ||
| arguments = self._field(item, "arguments", "{}") | ||
| if isinstance(arguments, str): | ||
| try: | ||
| parsed_arguments = json.loads(arguments) | ||
| except json.JSONDecodeError as exc: | ||
| logger.error("Failed to parse function arguments: %s", exc) | ||
| parsed_arguments = {} | ||
| else: | ||
| parsed_arguments = arguments |
Contributor
There was a problem hiding this comment.
suggestion: Responses API function call outputs and missing tool names/call IDs are not guarded against, which may cause subtle tool-calling issues.
In _parse_response, tool handling is limited to item_type == "function_call" and assumes non-empty name and call_id. The Responses API may also emit function_call_output or entries with missing/invalid call_id. Please either ignore or log unexpected tool item types (e.g., function_call_output) and add checks for missing/empty name and call_id so invalid tool calls are not added to LLMResponse.
Suggested implementation:
if item_type in ("function_call", "function_call_output") and tools is not None:
# Ignore unexpected tool item types (e.g., function_call_output) but log them for debugging.
if item_type != "function_call":
logger.warning("Ignoring unexpected tool item type in response: %s", item_type)
continue
# Validate that we have a usable tool name and call_id; otherwise, skip this tool call.
name = self._field(item, "name")
call_id = self._field(item, "call_id")
if not name or not call_id:
logger.warning(
"Skipping tool call with missing or empty name/call_id: name=%r, call_id=%r",
name,
call_id,
)
continue
arguments = self._field(item, "arguments", "{}")
if isinstance(arguments, str):
try:
parsed_arguments = json.loads(arguments)
except json.JSONDecodeError as exc:
logger.error("Failed to parse function arguments: %s", exc)
parsed_arguments = {}
else:
parsed_arguments = arguments
if parsed_arguments is None:- If later in this function you create tool call entries for
LLMResponseby re-readingnameorcall_idfromitem, you do not need to change that logic; the new validation already ensures only valid tool items reach that part of the code. - If you have or add metrics/structured logging, you may want to replace the
logger.warningcalls with your standard logging/telemetry helpers to ensure these invalid/ignored tool events are captured consistently.
Comment on lines
+50
to
+61
| def test_responses_provider_templates_are_independent_and_stateless(): | ||
| templates = CONFIG_METADATA_2["provider_group"]["metadata"]["provider"][ | ||
| "config_template" | ||
| ] | ||
|
|
||
| assert templates["OpenAI Responses"]["type"] == "openai_responses" | ||
| assert templates["OpenAI Responses"]["api_base"] == "https://api.openai.com/v1" | ||
| assert templates["DeepSeek Responses"]["type"] == "openai_responses" | ||
| assert templates["DeepSeek Responses"]["api_base"] == "https://api.deepseek.com/v1" | ||
| assert templates["xAI"]["type"] == "openai_responses" | ||
| assert templates["xAI"]["api_base"] == "https://api.x.ai/v1" | ||
| assert "xai_native_search" not in templates["xAI"] |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add explicit tests for the xAI chat-completions template and conditions to guard against regressions.
This test covers the new Responses-based xAI template, but CONFIG_METADATA_2 still defines the legacy xai_chat_completion template and xai_native_search flag. Please add assertions that the legacy "xAI Chat Completions" template is still present, has type == "xai_chat_completion", and that xai_native_search is scoped only to that template. This will better protect against regressions in the legacy search behavior during the Responses migration.
Suggested change
| def test_responses_provider_templates_are_independent_and_stateless(): | |
| templates = CONFIG_METADATA_2["provider_group"]["metadata"]["provider"][ | |
| "config_template" | |
| ] | |
| assert templates["OpenAI Responses"]["type"] == "openai_responses" | |
| assert templates["OpenAI Responses"]["api_base"] == "https://api.openai.com/v1" | |
| assert templates["DeepSeek Responses"]["type"] == "openai_responses" | |
| assert templates["DeepSeek Responses"]["api_base"] == "https://api.deepseek.com/v1" | |
| assert templates["xAI"]["type"] == "openai_responses" | |
| assert templates["xAI"]["api_base"] == "https://api.x.ai/v1" | |
| assert "xai_native_search" not in templates["xAI"] | |
| def test_responses_provider_templates_are_independent_and_stateless(): | |
| templates = CONFIG_METADATA_2["provider_group"]["metadata"]["provider"][ | |
| "config_template" | |
| ] | |
| # Responses-based templates | |
| assert templates["OpenAI Responses"]["type"] == "openai_responses" | |
| assert templates["OpenAI Responses"]["api_base"] == "https://api.openai.com/v1" | |
| assert "xai_native_search" not in templates["OpenAI Responses"] | |
| assert templates["DeepSeek Responses"]["type"] == "openai_responses" | |
| assert templates["DeepSeek Responses"]["api_base"] == "https://api.deepseek.com/v1" | |
| assert "xai_native_search" not in templates["DeepSeek Responses"] | |
| assert templates["xAI"]["type"] == "openai_responses" | |
| assert templates["xAI"]["api_base"] == "https://api.x.ai/v1" | |
| assert "xai_native_search" not in templates["xAI"] | |
| # Legacy xAI chat-completions template must remain present and correctly typed | |
| assert "xAI Chat Completions" in templates | |
| assert templates["xAI Chat Completions"]["type"] == "xai_chat_completion" | |
| # The legacy xai_native_search flag is scoped only to the chat-completions template | |
| assert "xai_native_search" in templates["xAI Chat Completions"] |
| return value.get(name, default) | ||
| return getattr(value, name, default) | ||
|
|
||
| def _convert_chat_messages_to_response_input( |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting small helper methods for message conversion, payload shaping, response parsing, and retry payload reconstruction to keep this adapter’s core methods linear and easier to maintain.
You can keep all current behavior while reducing complexity with a few small, focused helpers.
1. Reduce branching in _convert_chat_messages_to_response_input
Split out user/assistant content conversion and reasoning extraction; keep the main method as orchestration:
def _is_deepseek(self) -> bool:
host = (self.client.base_url.host or "").rstrip(".").lower()
return (
self.provider_config.get("provider") == "deepseek"
or host == "api.deepseek.com"
)
def _extract_reasoning_items_from_content(
self, content: list[dict], is_deepseek: bool
) -> list[dict]:
reasoning_items: list[dict] = []
for part in content:
if not isinstance(part, dict):
continue
if part.get("type") != "think":
continue
serialized_state = part.get("encrypted")
restored_items: list[dict] = []
if isinstance(serialized_state, str):
try:
state = json.loads(serialized_state)
except json.JSONDecodeError:
state = None
if (
isinstance(state, dict)
and state.get("type") == self._REASONING_STATE_TYPE
and isinstance(state.get("items"), list)
):
restored_items = [
item for item in state["items"] if isinstance(item, dict)
]
if restored_items:
reasoning_items.extend(restored_items)
elif is_deepseek and part.get("think"):
reasoning_items.append(
{
"type": "reasoning",
"content": [
{
"type": "reasoning_text",
"text": str(part["think"]),
}
],
"summary": [],
}
)
return reasoning_itemsThen simplify _convert_chat_messages_to_response_input by delegating:
def _convert_chat_messages_to_response_input(
self, messages: list[dict],
) -> list[dict]:
response_input: list[dict] = []
is_deepseek = self._is_deepseek()
for message in messages:
# tool/tool_call_output handling unchanged ...
role = message.get("role")
if role not in {"user", "assistant", "system", "developer"}:
continue
content = message.get("content")
reasoning_items: list[dict] = []
converted_content: str | list[dict] | None = None
if isinstance(content, list):
reasoning_items = self._extract_reasoning_items_from_content(
content, is_deepseek
)
# small helper can convert text/image/audio parts:
converted_content = self._convert_user_content_parts(content, role)
elif isinstance(content, str):
converted_content = content
elif content is not None:
converted_content = str(content)
response_input.extend(reasoning_items)
if converted_content not in (None, "", []):
response_input.append(
{
"type": "message",
"role": role,
"content": converted_content,
}
)
if role == "assistant":
self._append_tool_calls(response_input, message)
return response_inputYou already have the logic for tool calls; pulling it into _append_tool_calls keeps this method linear and easier to maintain.
2. Deduplicate _query / _query_stream payload shaping
The tools/extra_body handling is almost identical. A shared helper keeps both methods focused on transport:
def _prepare_responses_request(
self,
payloads: dict[str, Any],
tools: ToolSet | None,
) -> tuple[dict[str, Any], dict[str, Any]]:
payloads = dict(payloads) # shallow copy to avoid mutating caller
if tools:
response_tools = []
for tool in tools.openai_schema():
function = tool.get("function", {})
response_tools.append({"type": "function", **function})
if response_tools:
payloads["tools"] = response_tools
payloads["tool_choice"] = payloads.get("tool_choice", "auto")
extra_body: dict[str, Any] = {}
custom_extra_body = self.provider_config.get("custom_extra_body", {})
if isinstance(custom_extra_body, dict):
extra_body.update(custom_extra_body)
for key in list(payloads):
if key not in self.default_params:
extra_body[key] = payloads.pop(key)
max_tokens = extra_body.pop("max_tokens", None)
if max_tokens is not None and "max_output_tokens" not in extra_body:
extra_body["max_output_tokens"] = max_tokens
reasoning_effort = extra_body.pop("reasoning_effort", None)
if reasoning_effort is not None and "reasoning" not in extra_body:
extra_body["reasoning"] = {"effort": reasoning_effort}
for k in ("previous_response_id", "conversation", "store"):
extra_body.pop(k, None)
payloads.pop(k, None)
payloads["store"] = False
return payloads, extra_bodyThen _query and _query_stream can be reduced to:
async def _query(self, payloads: dict, tools: ToolSet | None, *, request_max_retries: int | None = None) -> LLMResponse:
payloads, extra_body = self._prepare_responses_request(payloads, tools)
response = await retry_provider_request(
"OpenAI Responses",
lambda: self.client.responses.create(
**payloads,
stream=False,
extra_body=extra_body,
),
max_attempts=request_max_retries,
)
# type check + parse as today...
async def _query_stream(self, payloads: dict, tools: ToolSet | None, *, request_max_retries: int | None = None) -> AsyncGenerator[LLMResponse, None]:
payloads, extra_body = self._prepare_responses_request(payloads, tools)
stream = await retry_provider_request(
"OpenAI Responses",
lambda: self.client.responses.create(
**payloads,
stream=True,
extra_body=extra_body,
),
max_attempts=request_max_retries,
)
# streaming event loop as today...This removes duplication and makes future changes to payload shaping much safer.
3. Isolate parsing responsibilities in _parse_response
You can keep _parse_response as the orchestrator and move the inner loops into helpers without altering behavior:
def _extract_output_and_reasoning(
self, response: Response,
) -> tuple[list[str], list[str], list[dict]]:
text_parts: list[str] = []
reasoning_parts: list[str] = []
serialized_reasoning_items: list[dict] = []
for item in self._field(response, "output", []) or []:
item_type = self._field(item, "type")
if item_type == "message":
for content in self._field(item, "content", []) or []:
content_type = self._field(content, "type")
if content_type == "output_text":
text_parts.append(str(self._field(content, "text", "")))
elif content_type == "refusal":
text_parts.append(str(self._field(content, "refusal", "")))
elif item_type == "reasoning":
# existing serialization + summary/content handling moved here
# ...
pass
return text_parts, reasoning_parts, serialized_reasoning_itemsThen _parse_response can do:
text_parts, reasoning_parts, serialized_items = self._extract_output_and_reasoning(response)
completion_text = "".join(text_parts)
if completion_text:
llm_response.result_chain = MessageChain().message(completion_text)
if reasoning_parts:
llm_response.reasoning_content = "\n".join(reasoning_parts)
if serialized_items:
llm_response.reasoning_signature = json.dumps(
{"type": self._REASONING_STATE_TYPE, "items": serialized_items},
ensure_ascii=False,
)Similarly, tool call extraction and usage can be moved into _extract_tool_calls and _extract_usage, shortening _parse_response and keeping each concern localized.
4. Simplify _handle_api_error payload reconstruction
To avoid duplicating the conversion logic and the “messages vs input” dance, introduce a helper that knows how to rebuild the Responses payload from the chat-format context:
def _rebuild_responses_payload_from_context(
self, payloads: dict, context_query: list[dict],
) -> dict:
retry_payloads = dict(payloads)
retry_payloads.pop("messages", None)
retry_payloads["input"] = self._convert_chat_messages_to_response_input(
context_query
)
retry_payloads["store"] = False
return retry_payloadsThen _handle_api_error becomes:
async def _handle_api_error(...):
compatibility_payloads = dict(payloads)
compatibility_payloads["messages"] = context_query
result = await super()._handle_api_error(
error,
compatibility_payloads,
context_query,
func_tool,
chosen_key,
available_api_keys,
retry_cnt,
max_retries,
image_fallback_used=image_fallback_used,
)
(
success,
chosen_key,
available_api_keys,
retry_payloads,
context_query,
func_tool,
image_fallback_used,
) = result
retry_payloads = self._rebuild_responses_payload_from_context(
retry_payloads, context_query
)
return (
success,
chosen_key,
available_api_keys,
retry_payloads,
context_query,
func_tool,
image_fallback_used,
)This keeps all current behavior but makes the retry/error path easier to reason about and less coupled to the internal structure of the payload dict.
This was referenced Aug 2, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
openai_responsesprovider adapter for OpenAI-compatible Responses endpointsstore: false, including reasoning and function-call itemsWhy
AstrBot only supported OpenAI-compatible Chat Completions, so Responses-first endpoints could not be configured independently without losing tool or reasoning history. OpenAI, DeepSeek, and xAI now expose compatible Responses endpoints, and xAI recommends Responses API over its legacy Chat Completions endpoint.
Impact
Existing Chat Completions provider configurations remain unchanged. New OpenAI and DeepSeek Responses providers can be selected independently, and newly created xAI providers use Responses API by default. All Responses calls remain stateless and resend the full local conversation history.
DeepSeek currently limits its Responses endpoint to supported Responses models such as
deepseek-v4-flash.Validation
uv run ruff format .uv run ruff check .uv run pytest -q tests/test_openai_source.py tests/test_openai_responses_source.py tests/test_fastapi_v1_dashboard.py— 157 passeduv run pytest -q tests/test_openai_responses_source.py— 8 passeduv run pytest -q tests— 2045 passed; 3 unrelated sandbox-cache failures caused by a locally installedkimi-datasourceskill leaking into test isolationSummary by Sourcery
Introduce a stateless OpenAI-compatible Responses provider and make xAI default to the Responses API while keeping legacy chat completions search opt-in.
New Features:
Enhancements:
Tests: