Bug: Cloud Agent _extract_user_message_text 返回字符串导致 API 400 错误
SDK 版本: qoder-agent-sdk 1.0.5 (Python)
平台: macOS / Python 3.12
文件: qoder_agent_sdk/_internal/cloud_agent.py
问题描述
使用 experimental_cloud_agent 调用 Cloud Agent 时，SDK 内部将用户消息的 content 从 content blocks 数组提取为纯字符串后再发送给 Cloud API，但 API 要求 content 必须是数组格式，导致 400 错误。
错误信息
CloudAgentApiError: Cloud Agent API request failed with status 400:
{"error":{"message":"Field 'content' must be an array of content blocks.","type":"invalid_request_error"}}
根因分析
调用链路如下：
第 1 步 — _to_user_message_stream() (line 1380) 正确地将字符串 prompt 转为 content blocks 数组：
# line 1380-1388
yield {
"type": "user",
"message": {
"role": "user",
"content": [{"type": "text", "text": prompt}], # ✅ 数组格式
},
}
第 2 步 — _extract_user_message_text() (line 1355) 将 content blocks 数组降级为纯字符串：
# line 1355-1370
def _extract_user_message_text(msg: dict[str, Any]) -> str: # 返回类型是 str
...
if isinstance(content, list):
return "\n".join(block["text"] ... ) # ❌ 把数组拼接成字符串
第 3 步 — _stream_input() (line 513) 用字符串作为 content 发送请求：
# line 510-513
content = _extract_user_message_text(msg) # 纯字符串
posted_events = await self._post_session_events(
[{"type": CLOUD_EVENT_USER_MESSAGE, "content": content}] # ❌ 发送字符串
)
Cloud API 期望的格式：
{"type": "user.message", "content": [{"type": "text", "text": "..."}]}
SDK 实际发送的格式：
{"type": "user.message", "content": "..."} // 字符串，被 API 拒绝
复现代码
from qoder_agent_sdk import QoderAgentOptions, access_token_from_env, query
import os
os.environ["QODER_PERSONAL_ACCESS_TOKEN"] = "<your-pat>"
os.environ["QODER_CLOUD_AGENT_ENVIRONMENT_ID"] = "<your-env-id>"
async for message in query(
prompt="Hello",
options=QoderAgentOptions(
auth=access_token_from_env(),
experimental_cloud_agent={
"agent": {"create": {"name": "test-agent", "model": "ultimate"}},
"session": {"create": {"environment_id": os.environ["QODER_CLOUD_AGENT_ENVIRONMENT_ID"]}},
},
),
):
print(message)
建议修复
方案 A：修改 _extract_user_message_text 返回 content blocks 数组：
def _extract_user_message_content(msg: dict[str, Any]) -> list[dict[str, Any]]:
message = msg.get("message")
if not isinstance(message, dict):
return [{"type": "text", "text": json.dumps(msg)}]
content = message.get("content")
if isinstance(content, str):
return [{"type": "text", "text": content}]
if isinstance(content, list):
return content
return [{"type": "text", "text": str(content)}]
方案 B：在 _stream_input 中包装字符串：
content = _extract_user_message_text(msg)
if isinstance(content, str):
content = [{"type": "text", "text": content}]
posted_events = await self._post_session_events(
[{"type": CLOUD_EVENT_USER_MESSAGE, "content": content}]
)
临时 Workaround
用户侧 monkey-patch：
from qoder_agent_sdk._internal import cloud_agent as _ca
def _patched_extract(msg):
message = msg.get("message")
if not isinstance(message, dict):
return [{"type": "text", "text": str(msg)}]
content = message.get("content")
if isinstance(content, str):
return [{"type": "text", "text": content}]
if isinstance(content, list):
return content
return [{"type": "text", "text": str(content)}]
_ca._extract_user_message_text = _patched_extract
