fix(xai): 迁移至 Responses API 以恢复联网搜索（migrate to the Responses API to r… - #9452
Open
Technistretron wants to merge 2 commits into
Open
fix(xai): 迁移至 Responses API 以恢复联网搜索（migrate to the Responses API to r…#9452Technistretron wants to merge 2 commits into
Technistretron wants to merge 2 commits into
Conversation
…estore web search） 自 2026 年 5 月起，xAI 官方 API 的调用方式发生变更，导致 AstrBot 中的 xAI 联网搜索无法正常启用，且无法再通过添加自定义请求键解决。通过 OpenRouter 调用 Grok 模型不受此问题影响。 本次修改更新了 xai_source.py，增加对 xAI Responses API 的支持，从而恢复官方 xAI API 的联网搜索功能。 该修改已在 Windows 版 AstrBot Desktop 上测试通过。使用时仍需在 xAI 模型配置中启用原有的“模型内置联网”选项。 Since May 2026, changes to the official xAI API have prevented AstrBot from enabling web search for xAI models. This can no longer be resolved by adding custom request keys, while Grok models accessed through OpenRouter remain unaffected. This change updates xai_source.py with support for the xAI Responses API, restoring web search when using the official xAI API. The implementation has been tested successfully with AstrBot Desktop on Windows. The existing built-in web search option must still be enabled in the xAI model configuration.
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 1 issue, and left some high level feedback:
- The new Responses conversion helpers (_convert_message_content, _convert_messages_to_input, _response_to_chat_completion) are fairly complex; consider factoring them into smaller, separately testable functions or a dedicated utility module to reduce cognitive load in ProviderXAI.
- In _query_stream, events are handled as Any with stringly-typed type checks; if the xAI SDK exposes typed stream event objects, wiring those through here would improve robustness and help catch shape changes at type-check time.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new Responses conversion helpers (_convert_message_content, _convert_messages_to_input, _response_to_chat_completion) are fairly complex; consider factoring them into smaller, separately testable functions or a dedicated utility module to reduce cognitive load in ProviderXAI.
- In _query_stream, events are handled as Any with stringly-typed type checks; if the xAI SDK exposes typed stream event objects, wiring those through here would improve robustness and help catch shape changes at type-check time.
## Individual Comments
### Comment 1
<location path="astrbot/core/provider/sources/xai_source.py" line_range="64" />
<code_context>
+ return value
+ return json.dumps(value, ensure_ascii=False, default=str)
+
+ def _convert_message_content(self, content: Any, role: str) -> Any:
+ """将 Chat message content 转换成 Responses input content。
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the message/content conversion and tool-merging logic into smaller helper functions to simplify control flow and make the adapter’s behavior easier to understand and modify.
A few focused extractions would significantly reduce complexity without changing behavior:
### 1. Flatten message/content conversion by role/part helpers
`_convert_message_content` and `_convert_messages_to_input` mix role-specific logic, error handling, and part-type branching. You can pull the per-part and per-role logic out to helpers so the main functions become simple loops.
Example refactor:
```python
def _convert_content_part(self, part: Any) -> dict[str, Any] | None:
if not isinstance(part, dict):
return {"type": "input_text", "text": str(part)}
part_type = part.get("type")
if part_type == "text":
return {"type": "input_text", "text": str(part.get("text", ""))}
if part_type == "image_url":
image_data = part.get("image_url", {})
if isinstance(image_data, str):
image_url, detail = image_data, None
elif isinstance(image_data, dict):
image_url = image_data.get("url")
detail = image_data.get("detail")
else:
image_url, detail = None, None
if not image_url:
raise ValueError("xAI Responses image input is missing image_url.")
image_part = {"type": "input_image", "image_url": image_url}
if detail:
image_part["detail"] = detail
return image_part
if part_type in {"audio_url", "input_audio"}:
raise ValueError(
"xAI Responses text models do not support AstrBot audio input.",
)
if part_type == "think":
return None
if part_type in {"input_text", "input_image", "input_file"}:
return copy.deepcopy(part)
raise ValueError(f"Unsupported xAI Responses content part: {part_type!r}")
def _convert_message_content(self, content: Any, role: str) -> Any:
if content is None:
return ""
if isinstance(content, str):
return content
if not isinstance(content, list):
return str(content)
converted_parts: list[dict[str, Any]] = []
for part in content:
converted = self._convert_content_part(part)
if converted is not None:
converted_parts.append(converted)
if role == "assistant" and all(
p.get("type") == "input_text" for p in converted_parts
):
return "".join(str(p.get("text", "")) for p in converted_parts)
return converted_parts
```
Then split `_convert_messages_to_input` by role to reduce nesting:
```python
def _convert_tool_message(self, message: dict[str, Any]) -> dict[str, Any]:
call_id = message.get("tool_call_id")
if not call_id:
raise ValueError("Tool result is missing tool_call_id.")
return {
"type": "function_call_output",
"call_id": call_id,
"output": self._to_json_string(message.get("content")),
}
def _convert_assistant_with_tool_calls(self, message: dict[str, Any]) -> list[dict[str, Any]]:
role = message["role"]
content = message.get("content")
tool_calls = message.get("tool_calls") or []
items: list[dict[str, Any]] = []
converted_content = self._convert_message_content(content, role)
if converted_content not in {"", None}:
items.append({"role": "assistant", "content": converted_content})
for tool_call in tool_calls:
function = self._get_value(tool_call, "function", {})
call_id = self._get_value(tool_call, "id")
name = self._get_value(function, "name")
arguments = self._get_value(function, "arguments", "{}")
if not call_id or not name:
raise ValueError("Assistant tool call is missing id or function name.")
items.append(
{
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": self._to_json_string(arguments),
}
)
return items
def _convert_messages_to_input(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
response_input: list[dict[str, Any]] = []
for message in messages:
role = message.get("role")
if role == "tool":
response_input.append(self._convert_tool_message(message))
elif role == "assistant" and (message.get("tool_calls") or []):
response_input.extend(self._convert_assistant_with_tool_calls(message))
elif role in {"system", "user", "assistant"}:
response_input.append(
{
"role": role,
"content": self._convert_message_content(
message.get("content"),
role,
),
}
)
else:
raise ValueError(f"Unsupported xAI Responses message role: {role!r}")
return response_input
```
This keeps behavior intact but makes the conversion rules much easier to follow and change.
### 2. Split tool merging into small, staged helpers
`_merge_tools` currently does collection, normalization, web-search injection, and deduplication in one go. Breaking that into small helpers makes each concern clearer:
```python
def _collect_tools(
self,
astrbot_tools: ToolSet | None,
custom_tools: Any,
) -> list[dict[str, Any]]:
tools: list[dict[str, Any]] = []
if astrbot_tools:
tools.extend(
self._normalize_tool(t) for t in astrbot_tools.get_func_desc_openai_style()
)
if custom_tools is not None:
if not isinstance(custom_tools, list):
raise ValueError("custom_extra_body.tools must be a list.")
tools.extend(self._normalize_tool(t) for t in custom_tools)
return tools
def _inject_web_search(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
if not self.provider_config.get("xai_native_search", False):
return tools
has_web_search = any(t.get("type") == "web_search" for t in tools)
if not has_web_search:
tools = tools + [{"type": "web_search"}]
return tools
def _deduplicate_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
deduped: list[dict[str, Any]] = []
seen: set[tuple[str, str]] = set()
for tool in tools:
tool_type = str(tool.get("type", ""))
identity = ("function", str(tool.get("name", ""))) if tool_type == "function" else ("builtin", tool_type)
if identity in seen:
continue
seen.add(identity)
deduped.append(tool)
return deduped
def _merge_tools(
self,
astrbot_tools: ToolSet | None,
custom_tools: Any,
) -> list[dict[str, Any]]:
tools = self._collect_tools(astrbot_tools, custom_tools)
tools = self._inject_web_search(tools)
return self._deduplicate_tools(tools)
```
This keeps all existing semantics (priority of AstrBot tools, web_search injection, deduplication) while making the behavior easier to audit and extend.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return value | ||
| return json.dumps(value, ensure_ascii=False, default=str) | ||
|
|
||
| def _convert_message_content(self, content: Any, role: str) -> Any: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the message/content conversion and tool-merging logic into smaller helper functions to simplify control flow and make the adapter’s behavior easier to understand and modify.
A few focused extractions would significantly reduce complexity without changing behavior:
1. Flatten message/content conversion by role/part helpers
_convert_message_content and _convert_messages_to_input mix role-specific logic, error handling, and part-type branching. You can pull the per-part and per-role logic out to helpers so the main functions become simple loops.
Example refactor:
def _convert_content_part(self, part: Any) -> dict[str, Any] | None:
if not isinstance(part, dict):
return {"type": "input_text", "text": str(part)}
part_type = part.get("type")
if part_type == "text":
return {"type": "input_text", "text": str(part.get("text", ""))}
if part_type == "image_url":
image_data = part.get("image_url", {})
if isinstance(image_data, str):
image_url, detail = image_data, None
elif isinstance(image_data, dict):
image_url = image_data.get("url")
detail = image_data.get("detail")
else:
image_url, detail = None, None
if not image_url:
raise ValueError("xAI Responses image input is missing image_url.")
image_part = {"type": "input_image", "image_url": image_url}
if detail:
image_part["detail"] = detail
return image_part
if part_type in {"audio_url", "input_audio"}:
raise ValueError(
"xAI Responses text models do not support AstrBot audio input.",
)
if part_type == "think":
return None
if part_type in {"input_text", "input_image", "input_file"}:
return copy.deepcopy(part)
raise ValueError(f"Unsupported xAI Responses content part: {part_type!r}")
def _convert_message_content(self, content: Any, role: str) -> Any:
if content is None:
return ""
if isinstance(content, str):
return content
if not isinstance(content, list):
return str(content)
converted_parts: list[dict[str, Any]] = []
for part in content:
converted = self._convert_content_part(part)
if converted is not None:
converted_parts.append(converted)
if role == "assistant" and all(
p.get("type") == "input_text" for p in converted_parts
):
return "".join(str(p.get("text", "")) for p in converted_parts)
return converted_partsThen split _convert_messages_to_input by role to reduce nesting:
def _convert_tool_message(self, message: dict[str, Any]) -> dict[str, Any]:
call_id = message.get("tool_call_id")
if not call_id:
raise ValueError("Tool result is missing tool_call_id.")
return {
"type": "function_call_output",
"call_id": call_id,
"output": self._to_json_string(message.get("content")),
}
def _convert_assistant_with_tool_calls(self, message: dict[str, Any]) -> list[dict[str, Any]]:
role = message["role"]
content = message.get("content")
tool_calls = message.get("tool_calls") or []
items: list[dict[str, Any]] = []
converted_content = self._convert_message_content(content, role)
if converted_content not in {"", None}:
items.append({"role": "assistant", "content": converted_content})
for tool_call in tool_calls:
function = self._get_value(tool_call, "function", {})
call_id = self._get_value(tool_call, "id")
name = self._get_value(function, "name")
arguments = self._get_value(function, "arguments", "{}")
if not call_id or not name:
raise ValueError("Assistant tool call is missing id or function name.")
items.append(
{
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": self._to_json_string(arguments),
}
)
return items
def _convert_messages_to_input(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
response_input: list[dict[str, Any]] = []
for message in messages:
role = message.get("role")
if role == "tool":
response_input.append(self._convert_tool_message(message))
elif role == "assistant" and (message.get("tool_calls") or []):
response_input.extend(self._convert_assistant_with_tool_calls(message))
elif role in {"system", "user", "assistant"}:
response_input.append(
{
"role": role,
"content": self._convert_message_content(
message.get("content"),
role,
),
}
)
else:
raise ValueError(f"Unsupported xAI Responses message role: {role!r}")
return response_inputThis keeps behavior intact but makes the conversion rules much easier to follow and change.
2. Split tool merging into small, staged helpers
_merge_tools currently does collection, normalization, web-search injection, and deduplication in one go. Breaking that into small helpers makes each concern clearer:
def _collect_tools(
self,
astrbot_tools: ToolSet | None,
custom_tools: Any,
) -> list[dict[str, Any]]:
tools: list[dict[str, Any]] = []
if astrbot_tools:
tools.extend(
self._normalize_tool(t) for t in astrbot_tools.get_func_desc_openai_style()
)
if custom_tools is not None:
if not isinstance(custom_tools, list):
raise ValueError("custom_extra_body.tools must be a list.")
tools.extend(self._normalize_tool(t) for t in custom_tools)
return tools
def _inject_web_search(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
if not self.provider_config.get("xai_native_search", False):
return tools
has_web_search = any(t.get("type") == "web_search" for t in tools)
if not has_web_search:
tools = tools + [{"type": "web_search"}]
return tools
def _deduplicate_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
deduped: list[dict[str, Any]] = []
seen: set[tuple[str, str]] = set()
for tool in tools:
tool_type = str(tool.get("type", ""))
identity = ("function", str(tool.get("name", ""))) if tool_type == "function" else ("builtin", tool_type)
if identity in seen:
continue
seen.add(identity)
deduped.append(tool)
return deduped
def _merge_tools(
self,
astrbot_tools: ToolSet | None,
custom_tools: Any,
) -> list[dict[str, Any]]:
tools = self._collect_tools(astrbot_tools, custom_tools)
tools = self._inject_web_search(tools)
return self._deduplicate_tools(tools)This keeps all existing semantics (priority of AstrBot tools, web_search injection, deduplication) while making the behavior easier to audit and extend.
防止计算哈希时报错TypeError: unhashable type: 'list'
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
1 participant
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.
Motivation / 修改动机
AstrBot's official xAI provider currently uses the Chat Completions API and injects the legacy
search_parametersfield when native search is enabled. However, xAI's current native Web Search is exposed through the Responses API using theweb_searchserver-side tool, so the existing “built-in web search” option no longer works with the official xAI API.This PR migrates the xAI-specific provider adapter to the Responses API while keeping AstrBot's existing internal Chat Completions-compatible interfaces unchanged. Grok models accessed through OpenRouter are not affected because they use a different provider path.
AstrBot 当前的官方 xAI Provider 使用 Chat Completions API，并在启用原生联网时注入旧版
search_parameters字段。由于 xAI 当前的原生 Web Search 需要通过 Responses API 的web_search服务端工具调用，现有的“模型内置联网”选项已无法通过官方 xAI API 正常工作。本 PR 将 xAI 专用 Provider 迁移至 Responses API，同时保留 AstrBot 内部现有的 Chat Completions 兼容接口。通过 OpenRouter 调用的 Grok 模型使用不同的 Provider 路径，因此不受本次修改影响。
Closes #7697
Refs #3888
Modifications / 改动点
Only
astrbot/core/provider/sources/xai_source.pyis modified.仅修改了
astrbot/core/provider/sources/xai_source.py。Migrated both streaming and non-streaming xAI requests to
client.responses.create().Preserved the existing
xai_chat_completionprovider ID to avoid invalidating existing user configurations.Converted AstrBot's Chat Completions-style message history into Responses API input items.
Added conversion for text, image input, function calls, and function call results.
Converted AstrBot function tools into the flattened Responses API tool format.
Merged AstrBot tools, custom tools, and the xAI
web_searchserver-side tool with deduplication.Added
web_searchautomatically whenxai_native_searchis enabled.Mapped token and reasoning parameters to their Responses API equivalents.
Added explicit errors for unsupported legacy Chat Completions parameters instead of silently ignoring them.
Converted Responses API output back into
ChatCompletionobjects so the existing AstrBot parser and plugin interfaces can continue to work.Added conversion for text, refusals, reasoning summaries, local function calls, token usage, and finish reasons.
Added streaming handling for text deltas, reasoning-summary deltas, completion events, and error events.
Continued using AstrBot's existing retry and response-parsing logic.
Set
store=falseby default because AstrBot continues to manage the complete conversation history locally.No new dependencies were introduced.
将 xAI 的流式和非流式请求迁移至
client.responses.create()。保留现有的
xai_chat_completionProvider ID，避免已有用户配置失效。将 AstrBot 的 Chat Completions 格式消息历史转换为 Responses API 输入。
支持文本、图片、函数调用及函数调用结果的格式转换。
将 AstrBot 函数工具转换为 Responses API 的扁平工具格式。
合并 AstrBot 本地工具、自定义工具和 xAI
web_search服务端工具，并进行去重。启用
xai_native_search时自动添加web_search工具。将 Token 和 Reasoning 参数映射至 Responses API 对应格式。
对不受支持的旧版 Chat Completions 参数明确报错，避免静默忽略。
将 Responses API 输出转换回
ChatCompletion，继续复用 AstrBot 现有的响应解析和插件接口。支持文本、拒绝信息、Reasoning Summary、本地函数调用、Token Usage 和结束状态的转换。
支持流式文本增量、Reasoning Summary 增量、完成事件和错误事件。
继续复用 AstrBot 现有的重试及响应解析逻辑。
默认使用
store=false，由 AstrBot 在本地继续维护完整对话历史。未引入任何新依赖。
This is NOT a breaking change. / 这不是一个破坏性变更。
Verification Steps / 验证步骤
Install the modified
xai_source.pyinto the AstrBot Desktop backend.Configure a model using the official xAI API endpoint and a valid xAI API key.
Verify that a normal conversation works when built-in web search is disabled.
Enable the existing built-in web search option in the xAI model configuration.
Ask a question that requires current information.
Verify that the request succeeds through the Responses API and returns a response using native web search.
将修改后的
xai_source.py替换到 AstrBot Desktop 后端。使用官方 xAI API 地址和有效的 xAI API Key 配置模型。
关闭模型内置联网，确认普通对话可以正常响应。
在 xAI 模型配置中启用原有的“模型内置联网”选项。
询问需要实时信息的问题。
确认请求通过 Responses API 成功完成，并能使用原生联网搜索返回结果。
Screenshots or Test Results / 运行截图或测试结果
Test environment:
xai_source.py测试环境：
xai_source.pyFull automated unit tests and cross-provider regression tests have not yet been completed.
尚未完成完整的自动化单元测试及跨 Provider 回归测试。
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
This change implements the existing proposal in 建议更新 xAI (Grok) 接口接入方式以支持原生 Web Search 功能 #7697. / 本修改实现了现有 Issue 建议更新 xAI (Grok) 接口接入方式以支持原生 Web Search 功能 #7697 中的方案。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Migrate the official xAI provider adapter from the Chat Completions API to the Responses API while preserving AstrBot’s existing ChatCompletion-based interfaces and retry/parse logic.
New Features:
web_searchserver-side tool into model requests when configured.Enhancements:
xai_chat_completionprovider ID so current user configurations remain valid.