poluzhao/api2cursor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

优化代码

Browse source
This commit is contained in:
h88782481 2026-03-09 19:43:51 +08:00
parent 202731df74
commit 96fbc4da80
15 changed files with 2768 additions and 1383 deletions
Download patch file Download diff file Expand all files Collapse all files

49
README.md
View file

@ -11,23 +11,33 @@ Cursor 根据模型名发送不同格式的请求：
| `claude-sonnet-*`、`glm-*` | `/v1/chat/completions` (OpenAI CC) |
| `gpt-*`、`claude-opus-*` | `/v1/responses` (OpenAI Responses) |
而中转站通常只支持 `/v1/chat/completions` 或 `/v1/messages`。
而中转站通常只支持 `/v1/chat/completions`、`/v1/messages` 或 `/v1/responses`。
本项目在中间做协议转换，**不管 Cursor 发什么格式，都能正确转发到中转站；不管中转站返回什么格式，都让 Cursor 能正确接收**。
## 架构
```
可以把这个项目理解成“三种入口协议 + 三种上游后端协议”的协议桥：
```text
Cursor API 2 Cursor 中转站
│ │ │
├─ /v1/chat/completions ──→ chat.py ─┬─ openai 后端 ────────→ /v1/chat/completions
│ └─ anthropic 后端 ────→ /v1/messages
│ │
├─ /v1/responses ──────→ responses.py → 转为 CC → 同上 → 转回 Responses
│ │
└─ /v1/messages ───────→ messages.py → 直接透传 ────────────→ /v1/messages
├─ /v1/chat/completions ─────→ chat.py ─────┬─ openai 后端 ─────────→ /v1/chat/completions
│ ├─ anthropic 后端 ─────→ /v1/messages
│ └─ responses 后端 ─────→ /v1/responses
│
├─ /v1/responses ────────────→ responses.py ─┬─ openai 后端 ───────→ /v1/chat/completions
│ ├─ anthropic 后端 ───→ /v1/messages
│ └─ responses 后端 ───→ /v1/responses
│
└─ /v1/messages ─────────────→ messages.py ─────────────────────────→ /v1/messages
```
其中：
- `chat.py` 负责接住 Cursor 的 Chat Completions 请求，并根据模型映射决定发往哪种后端协议
- `responses.py` 负责接住 Cursor 的 Responses 请求，并在需要时做 `Responses ↔ CC` 或 `Responses ↔ Messages` 桥接
- `messages.py` 负责 Anthropic 原生消息的直通场景
## 快速开始
### 直接运行
@ -70,11 +80,13 @@ docker compose up -d
- **Cursor 模型名** — 在 Cursor 自定义模型中填入的名称
- **上游模型名** — 发送到中转站的实际模型名
- **后端类型** — `openai` (CC 格式) / `anthropic` (Messages 格式) / `auto` (自动检测)
- **后端类型** — `openai` (CC 格式) / `anthropic` (Messages 格式) / `responses` (Responses 格式) / `auto` (自动检测)
- **自定义地址/密钥** — 可选，覆盖全局设置，实现分流到不同中转站
**示例**：在 Cursor 中添加 `claude-sonnet-4-5-20250929`，映射到上游 `gpt-5.3-codex`，后端选 `openai`。Cursor 会用 CC 格式发送请求，代理直接转发到中转站的 `/v1/chat/completions`。
如果你的中转站只支持 `/v1/responses`，可以把后端类型选成 `responses`。此时代理会把 Cursor 发来的请求转换或透传为 Responses 格式，再发往中转站的 `/v1/responses`。
> **提示**：使用 Claude 风格的模型名（如 `claude-sonnet-4-5-20250929`）可以让 Cursor 显示思考过程（thinking）。
### 在 Cursor 中配置
@ -86,22 +98,23 @@ docker compose up -d
## 项目结构
```
```text
api2cursor/
├── start.py # 启动入口
├── app.py # Flask 应用工厂
├── config.py # 环境变量配置
├── settings.py # 持久化配置管理
├── routes/ # 路由层
├── routes/ # 路由层：按对外 API 入口拆分
│ ├── chat.py # /v1/chat/completions
│ ├── responses.py # /v1/responses
│ ├── messages.py # /v1/messages (透传)
│ └── admin.py # 管理面板 + API
├── adapters/ # 适配层（格式转换）
│ ├── openai_anthropic.py# CC ↔ Messages 双向转换
│ ├── openai_fixer.py # OpenAI 请求/响应修复
│ └── responses_adapter.py# Responses ↔ CC 双向转换
├── utils/ # 工具层
│ ├── messages.py # /v1/messages（透传）
│ ├── admin.py # 管理面板 + API
│ └── common.py # 路由公共上下文、日志与 SSE 辅助
├── adapters/ # 适配层：按协议桥接职责拆分
│ ├── cc_anthropic_adapter.py # Chat Completions ↔ Anthropic Messages
│ ├── openai_compat_fixer.py # OpenAI / Chat Completions 兼容修复
│ └── responses_cc_adapter.py # Responses ↔ Chat Completions + 原生 Responses 流桥接
├── utils/ # 通用工具层
│ ├── http.py # 请求转发、SSE 解析
│ ├── tool_fixer.py # 工具参数修复
│ └── think_tag.py # <think> 标签提取

561
adapters/cc_anthropic_adapter.py Normal file
View file

@ -0,0 +1,561 @@
"""OpenAI Chat Completions ↔ Anthropic Messages 格式转换
这个模块是项目里最核心的协议桥之一，负责在两套主流对话协议之间做双向适配：
- 请求方向：OpenAI Chat Completions → Anthropic Messages
- 响应方向：Anthropic Messages → OpenAI Chat Completions
- 流式方向：Anthropic SSE 事件 → OpenAI Chat Completions chunk
这里的代码看起来会比普通字段映射更重，是因为它不仅要做字段重命名，还要处理：
- system 消息上提
- tool_calls / tool_use 双向映射
- tool 消息 / tool_result 双向映射
- 图片块转换
- 思考内容与流式工具参数的时序保留
"""
from __future__ import annotations
import json
from typing import Any
from utils.http import gen_id
from utils.tool_fixer import fix_anthropic_tool_use, normalize_args, repair_str_replace_args
JsonDict = dict[str, Any]
# Anthropic stop_reason → OpenAI finish_reason
_STOP_REASON_MAP = {
'end_turn': 'stop',
'max_tokens': 'length',
'tool_use': 'tool_calls',
'stop_sequence': 'stop',
}
# ═══════════════════════════════════════════════════════════
# 请求转换: CC → Messages
# ═══════════════════════════════════════════════════════════
def cc_to_messages_request(payload: JsonDict) -> JsonDict:
"""将 OpenAI Chat Completions 请求转换为 Anthropic Messages 请求。
这一步不是简单替换字段名，而是主动把 OpenAI 世界中的几类特殊语义映射到
Anthropic 世界：
- `system` 消息提取到顶层 `system`
- assistant 的 `tool_calls` 变成 `tool_use` 内容块
- `tool` 角色消息变成 user 侧的 `tool_result` 内容块
另外，这里会把相邻同角色消息做合并，因为 Anthropic 对消息角色交替的要求更严格。
"""
messages = payload.get('messages', [])
anthropic_messages: list[JsonDict] = []
system_parts: list[str] = []
for message in messages:
converted, system_text = _convert_request_message(message)
if system_text is not None:
system_parts.append(system_text)
continue
if converted is not None:
anthropic_messages.append(converted)
anthropic_messages = _merge_same_role(anthropic_messages)
return _build_messages_request(payload, anthropic_messages, system_parts)
# ═══════════════════════════════════════════════════════════
# 非流式响应转换: Messages → CC
# ═══════════════════════════════════════════════════════════
def messages_to_cc_response(data: JsonDict, request_id: str | None = None) -> JsonDict:
"""将 Anthropic Messages 非流式响应转换为 OpenAI CC 响应。"""
request_id = request_id or gen_id('chatcmpl-')
data = fix_anthropic_tool_use(data)
content_text, reasoning_text, tool_calls = _collect_response_parts(data.get('content', []))
message = _build_cc_message(content_text, reasoning_text, tool_calls)
usage = data.get('usage', {})
return {
'id': request_id,
'object': 'chat.completion',
'model': data.get('model', 'claude'),
'choices': [{
'index': 0,
'message': message,
'finish_reason': _STOP_REASON_MAP.get(data.get('stop_reason', 'end_turn'), 'stop'),
}],
'usage': _build_cc_usage(
input_tokens=usage.get('input_tokens', 0),
output_tokens=usage.get('output_tokens', 0),
),
}
# ═══════════════════════════════════════════════════════════
# 流式响应转换: Anthropic SSE → CC chunks
# ═══════════════════════════════════════════════════════════
class AnthropicStreamConverter:
"""将 Anthropic SSE 事件逐个转换为 OpenAI Chat Completions chunk。
之所以做成有状态转换器，而不是单纯的函数映射，是因为 Anthropic 的流式工具调用
会把名字、参数、结束信号拆散在多个事件中，而 OpenAI chunk 语义要求我们按顺序
组装出连续的 `tool_calls` 增量。
这个类主要维护三类状态：
1. 当前请求的 chunk ID
2. 当前工具调用的索引位置
3. 输入 / 输出令牌统计
最终目标是把 Anthropic 的事件流稳定映射成 Cursor 能直接消费的 CC chunk 流。
"""
def __init__(self, request_id: str | None = None):
self._id = request_id or gen_id('chatcmpl-')
self._tool_index = -1
self._input_tokens = 0
self._output_tokens = 0
def process_event(self, event_type: str, event_data: JsonDict) -> list[str]:
"""处理单个 Anthropic SSE 事件。
调用方会按事件顺序不断喂入 event/data，这里根据事件类型拆成一个或多个 CC chunk
字符串，交给上层直接作为 SSE data 发送给 Cursor。
"""
if event_type == 'message_start':
return self._handle_message_start(event_data)
if event_type == 'content_block_start':
return self._handle_content_block_start(event_data)
if event_type == 'content_block_delta':
return self._handle_content_block_delta(event_data)
if event_type == 'message_delta':
return self._handle_message_delta(event_data)
return []
def _handle_message_start(self, event_data: JsonDict) -> list[str]:
"""处理消息开始事件，产出 assistant 角色起始 chunk。
这个起始 chunk 很重要，因为 Cursor 侧通常会依赖首个带 role 的 chunk 来初始化
当前 assistant 消息。
"""
message = event_data.get('message', {})
self._input_tokens = message.get('usage', {}).get('input_tokens', 0)
chunk = self._make_chunk(delta={'role': 'assistant', 'content': ''})
if message.get('model'):
chunk['model'] = message['model']
return [self._dump_chunk(chunk)]
def _handle_content_block_start(self, event_data: JsonDict) -> list[str]:
"""处理内容块开始事件。
目前这里只需要显式处理 `tool_use`，因为文本和 thinking 的真正内容都在后续 delta
事件里；而 tool_use 需要先开一个空 arguments 的 tool_call 槽位。
"""
block = event_data.get('content_block', {})
if block.get('type') != 'tool_use':
return []
self._tool_index += 1
return [self._dump_chunk(self._make_chunk(delta={
'tool_calls': [{
'index': self._tool_index,
'id': block.get('id', gen_id('toolu_')),
'type': 'function',
'function': {
'name': block.get('name', ''),
'arguments': '',
},
}]
}))]
def _handle_content_block_delta(self, event_data: JsonDict) -> list[str]:
"""处理内容块增量事件。
Anthropic 会把文本、思考内容、工具参数拆成不同 delta 类型，这里要分别映射成
OpenAI chunk 里的 `content`、`reasoning_content` 和 `tool_calls.function.arguments`。
"""
delta = event_data.get('delta', {})
delta_type = delta.get('type', '')
if delta_type == 'text_delta' and delta.get('text'):
return [self._dump_chunk(self._make_chunk(delta={'content': delta['text']}))]
if delta_type == 'thinking_delta' and delta.get('thinking'):
return [self._dump_chunk(self._make_chunk(delta={'reasoning_content': delta['thinking']}))]
if delta_type == 'input_json_delta' and delta.get('partial_json'):
return [self._dump_chunk(self._make_chunk(delta={
'tool_calls': [{
'index': self._tool_index,
'function': {'arguments': delta['partial_json']},
}]
}))]
return []
def _handle_message_delta(self, event_data: JsonDict) -> list[str]:
"""处理消息收尾事件，补出 finish_reason 和 usage。
当 Anthropic 发出 `message_delta` 时，说明这一轮 assistant 输出已经收束，
这里会统一生成最后一个带 usage 的收尾 chunk。
"""
delta = event_data.get('delta', {})
usage = event_data.get('usage', {})
self._output_tokens = usage.get('output_tokens', 0)
chunk = self._make_chunk(
delta={},
finish_reason=_STOP_REASON_MAP.get(delta.get('stop_reason', ''), 'stop'),
)
chunk['usage'] = _build_cc_usage(
input_tokens=self._input_tokens,
output_tokens=self._output_tokens,
)
return [self._dump_chunk(chunk)]
def _make_chunk(self, delta: JsonDict, finish_reason: str | None = None) -> JsonDict:
"""构造标准 OpenAI Chat Completions chunk 对象。"""
choice: JsonDict = {'index': 0, 'delta': delta}
if finish_reason:
choice['finish_reason'] = finish_reason
return {
'id': self._id,
'object': 'chat.completion.chunk',
'model': 'claude',
'choices': [choice],
}
@staticmethod
def _dump_chunk(chunk: JsonDict) -> str:
"""统一序列化 chunk，方便上层直接写入 SSE data。"""
return json.dumps(chunk)
# ═══════════════════════════════════════════════════════════
# 请求转换辅助
# ═══════════════════════════════════════════════════════════
def _convert_request_message(message: Any) -> tuple[JsonDict | None, str | None]:
"""将单条 OpenAI 消息转换为 Anthropic 消息或 system 文本。"""
if not isinstance(message, dict):
return None, None
role = message.get('role', '')
content = message.get('content', '')
if role == 'system':
return None, _flatten_text(content)
if role == 'tool':
return _convert_tool_role_message(message), None
anthropic_role = 'assistant' if role == 'assistant' else 'user'
anthropic_content = _convert_content(message)
if role == 'assistant' and 'tool_calls' in message:
anthropic_content = _append_tool_use_blocks(anthropic_content, message.get('tool_calls', []))
if not anthropic_content and anthropic_content != 0:
return None, None
return {'role': anthropic_role, 'content': anthropic_content}, None
def _convert_tool_role_message(message: JsonDict) -> JsonDict | None:
"""将 OpenAI 的 tool 角色消息转换为 Anthropic 的 tool_result 内容块。"""
content = message.get('content', '')
text = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
anthropic_content = [{
'type': 'tool_result',
'tool_use_id': message.get('tool_call_id', ''),
'content': text,
}]
if not anthropic_content:
return None
return {'role': 'user', 'content': anthropic_content}
def _append_tool_use_blocks(content: Any, tool_calls: list[Any]) -> list[JsonDict]:
"""把 OpenAI assistant.tool_calls 追加成 Anthropic tool_use 内容块。"""
blocks = _to_blocks(content)
for tool_call in tool_calls:
if not isinstance(tool_call, dict):
continue
function_data = tool_call.get('function', {})
blocks.append({
'type': 'tool_use',
'id': tool_call.get('id', gen_id('toolu_')),
'name': function_data.get('name', ''),
'input': _parse_tool_arguments(function_data.get('arguments', '{}')),
})
return blocks
def _build_messages_request(
payload: JsonDict,
anthropic_messages: list[JsonDict],
system_parts: list[str],
) -> JsonDict:
"""组装最终的 Anthropic Messages 请求体。"""
result: JsonDict = {
'model': payload.get('model', 'claude-sonnet-4-20250514'),
'messages': anthropic_messages,
# 沿用项目当前策略：未设置或设置过小都兜底到 8192，避免上游因默认值过小过早截断。
'max_tokens': max(payload.get('max_tokens') or 8192, 8192),
}
if system_parts:
result['system'] = '\n\n'.join(system_parts)
if 'tools' in payload:
result['tools'] = _convert_tools(payload['tools'])
for key in ('temperature', 'top_p', 'stream'):
if key in payload:
result[key] = payload[key]
return result
# ═══════════════════════════════════════════════════════════
# 非流式响应转换辅助
# ═══════════════════════════════════════════════════════════
def _collect_response_parts(content_blocks: Any) -> tuple[str, str, list[JsonDict]]:
"""从 Anthropic content 块中提取文本、思考内容和工具调用。"""
content_text = ''
reasoning_text = ''
tool_calls: list[JsonDict] = []
if not isinstance(content_blocks, list):
return content_text, reasoning_text, tool_calls
for block in content_blocks:
if not isinstance(block, dict):
continue
block_type = block.get('type', '')
if block_type == 'text':
content_text += block.get('text', '')
elif block_type == 'thinking':
reasoning_text += block.get('thinking', '')
elif block_type == 'tool_use':
tool_calls.append(_convert_tool_use_block(block, index=len(tool_calls)))
return content_text, reasoning_text, tool_calls
def _convert_tool_use_block(block: JsonDict, *, index: int) -> JsonDict:
"""将 Anthropic 的 tool_use 块转换为 OpenAI tool_call。"""
tool_name = block.get('name', '')
input_data = block.get('input', {})
if isinstance(input_data, dict):
input_data = normalize_args(input_data)
input_data = repair_str_replace_args(tool_name, input_data)
arguments_text = json.dumps(input_data, ensure_ascii=False)
else:
arguments_text = str(input_data)
return {
'index': index,
'id': block.get('id', gen_id('toolu_')),
'type': 'function',
'function': {
'name': tool_name,
'arguments': arguments_text,
},
}
def _build_cc_message(content_text: str, reasoning_text: str, tool_calls: list[JsonDict]) -> JsonDict:
"""构造 OpenAI CC 响应中的 assistant message。"""
message: JsonDict = {
'role': 'assistant',
'content': content_text or None,
}
if reasoning_text:
message['reasoning_content'] = reasoning_text
if tool_calls:
message['tool_calls'] = tool_calls
return message
def _build_cc_usage(*, input_tokens: int, output_tokens: int) -> JsonDict:
"""将 Anthropic usage 字段映射为 OpenAI usage。"""
return {
'prompt_tokens': input_tokens,
'completion_tokens': output_tokens,
'total_tokens': input_tokens + output_tokens,
}
# ═══════════════════════════════════════════════════════════
# 通用辅助
# ═══════════════════════════════════════════════════════════
def _parse_tool_arguments(arguments: Any) -> Any:
"""将 tool_call.arguments 尽量解析为对象，供 Anthropic tool_use.input 使用。
Anthropic 的 `tool_use.input` 天然期望对象结构；如果这里直接保留原始字符串，
后续上游会把它当普通文本而不是工具参数对象。
"""
if not isinstance(arguments, str):
return arguments if arguments is not None else {}
try:
return json.loads(arguments)
except json.JSONDecodeError:
return {}
def _flatten_text(content: Any) -> str:
"""将 content 扁平化为纯文本，主要用于 system 消息上提。"""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for part in content:
if isinstance(part, str):
parts.append(part)
elif isinstance(part, dict) and part.get('type') == 'text':
parts.append(part.get('text', ''))
return '\n'.join(parts)
return str(content)
def _convert_content(message: JsonDict) -> Any:
"""将 OpenAI 消息的 content 字段转换为 Anthropic 内容格式。"""
content = message.get('content', '')
if content is None:
return ''
if isinstance(content, str):
return content
if not isinstance(content, list):
return str(content)
blocks: list[JsonDict] = []
for part in content:
converted = _convert_content_part(part)
if converted is not None:
blocks.append(converted)
return blocks
def _convert_content_part(part: Any) -> JsonDict | None:
"""将单个 OpenAI content part 转为 Anthropic block。"""
if isinstance(part, str):
return {'type': 'text', 'text': part}
if not isinstance(part, dict):
return None
part_type = part.get('type', '')
if part_type == 'text':
return {'type': 'text', 'text': part.get('text', '')}
if part_type == 'image_url':
return _convert_image(part)
if part_type in ('tool_use', 'tool_result'):
return part
return None
def _convert_image(part: JsonDict) -> JsonDict:
"""将 OpenAI image_url 格式转换为 Anthropic image 格式。"""
url_data = part.get('image_url', {})
url = url_data.get('url', '') if isinstance(url_data, dict) else str(url_data)
if url.startswith('data:'):
media_type, _, base64_data = url.partition(';base64,')
return {
'type': 'image',
'source': {
'type': 'base64',
'media_type': media_type.replace('data:', '') or 'image/png',
'data': base64_data,
},
}
return {
'type': 'image',
'source': {
'type': 'url',
'url': url,
},
}
def _convert_tools(tools: Any) -> list[JsonDict]:
"""将 OpenAI tools 转为 Anthropic tools 格式。
这里兼容两种常见输入：
- 标准 OpenAI `{"type": "function", "function": {...}}`
- Cursor 常见的扁平工具格式 `{"name": ..., "input_schema": ...}`
"""
if not isinstance(tools, list):
return []
result: list[JsonDict] = []
for tool in tools:
converted = _convert_tool_definition(tool)
if converted is not None:
result.append(converted)
return result
def _convert_tool_definition(tool: Any) -> JsonDict | None:
"""转换单个工具定义。"""
if not isinstance(tool, dict):
return None
if tool.get('type') == 'function' and 'function' in tool:
function_data = tool['function']
return {
'name': function_data.get('name', ''),
'description': function_data.get('description', ''),
'input_schema': function_data.get('parameters', {'type': 'object', 'properties': {}}),
}
if 'name' in tool and 'input_schema' in tool:
return {
'name': tool.get('name', ''),
'description': tool.get('description', ''),
'input_schema': tool.get('input_schema', {'type': 'object', 'properties': {}}),
}
return None
def _to_blocks(content: Any) -> list[JsonDict]:
"""将内容统一转换成 block 列表。"""
if isinstance(content, str):
return [{'type': 'text', 'text': content}] if content else []
if isinstance(content, list):
return list(content)
return [{'type': 'text', 'text': str(content)}] if content else []
def _merge_same_role(messages: list[JsonDict]) -> list[JsonDict]:
"""合并相邻同角色消息。
Anthropic 要求消息角色严格交替，而 OpenAI/调用方不一定遵守这一点。
这里仅合并“相邻同角色”消息，以最小改动满足 Anthropic 约束，同时尽量保留
原本的消息顺序和内容块排列。
"""
if not messages:
return messages
merged = [messages[0]]
for message in messages[1:]:
if message['role'] == merged[-1]['role']:
previous_blocks = _to_blocks(merged[-1]['content'])
current_blocks = _to_blocks(message['content'])
merged[-1]['content'] = previous_blocks + current_blocks
else:
merged.append(message)
return merged

350
adapters/openai_anthropic.py
View file

@ -1,350 +0,0 @@
"""OpenAI Chat Completions ↔ Anthropic Messages 格式转换
请求方向: CC → Messages（Cursor 的 CC 请求转为 Anthropic 格式发给上游）
响应方向: Messages → CC（上游 Anthropic 响应转为 CC 格式返回给 Cursor）
包含非流式和流式两种转换。
"""
import json
import uuid
import logging
from utils.tool_fixer import normalize_args, repair_str_replace_args, fix_anthropic_tool_use
from utils.http import gen_id
logger = logging.getLogger(__name__)
# Anthropic stop_reason → OpenAI finish_reason
_STOP_REASON_MAP = {
'end_turn': 'stop',
'max_tokens': 'length',
'tool_use': 'tool_calls',
'stop_sequence': 'stop',
}
# ═══════════════════════════════════════════════════════════
# 请求转换: CC → Messages
# ═══════════════════════════════════════════════════════════
def cc_to_messages_request(payload):
"""将 OpenAI CC 格式请求转换为 Anthropic Messages 格式"""
messages = payload.get('messages', [])
anthropic_msgs = []
system_parts = []
for msg in messages:
role = msg.get('role', '')
content = msg.get('content', '')
# system 消息提取到顶层
if role == 'system':
system_parts.append(_flatten_text(content))
continue
anthropic_role = 'assistant' if role == 'assistant' else 'user'
anthropic_content = _convert_content(msg)
# assistant 的 tool_calls → tool_use content blocks
if role == 'assistant' and 'tool_calls' in msg:
blocks = _to_blocks(anthropic_content)
for tc in msg['tool_calls']:
func = tc.get('function', {})
arguments = func.get('arguments', '{}')
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except json.JSONDecodeError:
arguments = {}
blocks.append({
'type': 'tool_use',
'id': tc.get('id', f'toolu_{uuid.uuid4().hex[:24]}'),
'name': func.get('name', ''),
'input': arguments,
})
anthropic_content = blocks
# tool 角色 → user + tool_result
if role == 'tool':
text = content if isinstance(content, str) else json.dumps(content)
anthropic_content = [{
'type': 'tool_result',
'tool_use_id': msg.get('tool_call_id', ''),
'content': text,
}]
anthropic_role = 'user'
if not anthropic_content and anthropic_content != 0:
continue
anthropic_msgs.append({'role': anthropic_role, 'content': anthropic_content})
# Anthropic 要求角色必须交替
anthropic_msgs = _merge_same_role(anthropic_msgs)
result = {
'model': payload.get('model', 'claude-sonnet-4-20250514'),
'messages': anthropic_msgs,
'max_tokens': max(payload.get('max_tokens') or 8192, 8192),
}
if system_parts:
result['system'] = '\n\n'.join(system_parts)
if 'tools' in payload:
result['tools'] = _convert_tools(payload['tools'])
for key in ('temperature', 'top_p', 'stream'):
if key in payload:
result[key] = payload[key]
return result
# ═══════════════════════════════════════════════════════════
# 非流式响应转换: Messages → CC
# ═══════════════════════════════════════════════════════════
def messages_to_cc_response(data, request_id=None):
"""将 Anthropic Messages 响应转换为 OpenAI CC 格式"""
request_id = request_id or gen_id('chatcmpl-')
data = fix_anthropic_tool_use(data)
content_text = ''
reasoning = ''
tool_calls = []
for block in data.get('content', []):
if not isinstance(block, dict):
continue
btype = block.get('type', '')
if btype == 'text':
content_text += block.get('text', '')
elif btype == 'thinking':
reasoning += block.get('thinking', '')
elif btype == 'tool_use':
args = block.get('input', {})
if isinstance(args, dict):
args = normalize_args(args)
args = repair_str_replace_args(block.get('name', ''), args)
tool_calls.append({
'index': len(tool_calls),
'id': block.get('id', f'toolu_{uuid.uuid4().hex[:24]}'),
'type': 'function',
'function': {
'name': block.get('name', ''),
'arguments': json.dumps(args, ensure_ascii=False) if isinstance(args, dict) else str(args),
},
})
stop_reason = data.get('stop_reason', 'end_turn')
message = {'role': 'assistant', 'content': content_text or None}
if reasoning:
message['reasoning_content'] = reasoning
if tool_calls:
message['tool_calls'] = tool_calls
usage = data.get('usage', {})
return {
'id': request_id,
'object': 'chat.completion',
'model': data.get('model', 'claude'),
'choices': [{
'index': 0,
'message': message,
'finish_reason': _STOP_REASON_MAP.get(stop_reason, 'stop'),
}],
'usage': {
'prompt_tokens': usage.get('input_tokens', 0),
'completion_tokens': usage.get('output_tokens', 0),
'total_tokens': usage.get('input_tokens', 0) + usage.get('output_tokens', 0),
},
}
# ═══════════════════════════════════════════════════════════
# 流式响应转换: Anthropic SSE → CC chunks
# ═══════════════════════════════════════════════════════════
class AnthropicStreamConverter:
"""将 Anthropic SSE 事件逐个转换为 OpenAI CC 流式 chunk"""
def __init__(self, request_id=None):
self._id = request_id or gen_id('chatcmpl-')
self._tool_index = -1
self._input_tokens = 0
self._output_tokens = 0
def process_event(self, event_type, event_data):
"""处理一个 Anthropic SSE 事件，返回 CC chunk JSON 字符串列表"""
chunks = []
if event_type == 'message_start':
msg = event_data.get('message', {})
self._input_tokens = msg.get('usage', {}).get('input_tokens', 0)
chunk = self._make_chunk(delta={'role': 'assistant', 'content': ''})
if msg.get('model'):
chunk['model'] = msg['model']
chunks.append(json.dumps(chunk))
elif event_type == 'content_block_start':
block = event_data.get('content_block', {})
if block.get('type') == 'tool_use':
self._tool_index += 1
chunks.append(json.dumps(self._make_chunk(delta={
'tool_calls': [{
'index': self._tool_index,
'id': block.get('id', f'toolu_{uuid.uuid4().hex[:24]}'),
'type': 'function',
'function': {'name': block.get('name', ''), 'arguments': ''},
}]
})))
elif event_type == 'content_block_delta':
delta = event_data.get('delta', {})
dtype = delta.get('type', '')
if dtype == 'text_delta' and delta.get('text'):
chunks.append(json.dumps(self._make_chunk(
delta={'content': delta['text']})))
elif dtype == 'thinking_delta' and delta.get('thinking'):
chunks.append(json.dumps(self._make_chunk(
delta={'reasoning_content': delta['thinking']})))
elif dtype == 'input_json_delta' and delta.get('partial_json'):
chunks.append(json.dumps(self._make_chunk(delta={
'tool_calls': [{
'index': self._tool_index,
'function': {'arguments': delta['partial_json']},
}]
})))
elif event_type == 'message_delta':
delta = event_data.get('delta', {})
usage = event_data.get('usage', {})
self._output_tokens = usage.get('output_tokens', 0)
finish = _STOP_REASON_MAP.get(delta.get('stop_reason', ''), 'stop')
chunk = self._make_chunk(delta={}, finish_reason=finish)
chunk['usage'] = {
'prompt_tokens': self._input_tokens,
'completion_tokens': self._output_tokens,
'total_tokens': self._input_tokens + self._output_tokens,
}
chunks.append(json.dumps(chunk))
return chunks
def _make_chunk(self, delta, finish_reason=None):
choice = {'index': 0, 'delta': delta}
if finish_reason:
choice['finish_reason'] = finish_reason
return {
'id': self._id,
'object': 'chat.completion.chunk',
'model': 'claude',
'choices': [choice],
}
# ═══════════════════════════════════════════════════════════
# 内部辅助函数
# ═══════════════════════════════════════════════════════════
def _flatten_text(content):
"""将 content 扁平化为纯文本"""
if isinstance(content, str):
return content
if isinstance(content, list):
parts = []
for p in content:
if isinstance(p, str):
parts.append(p)
elif isinstance(p, dict) and p.get('type') == 'text':
parts.append(p.get('text', ''))
return '\n'.join(parts)
return str(content)
def _convert_content(msg):
"""将 OpenAI 消息的 content 字段转为 Anthropic 格式"""
content = msg.get('content', '')
if content is None:
return ''
if isinstance(content, str):
return content
if isinstance(content, list):
blocks = []
for part in content:
if isinstance(part, str):
blocks.append({'type': 'text', 'text': part})
elif isinstance(part, dict):
ptype = part.get('type', '')
if ptype == 'text':
blocks.append({'type': 'text', 'text': part.get('text', '')})
elif ptype == 'image_url':
blocks.append(_convert_image(part))
elif ptype in ('tool_use', 'tool_result'):
blocks.append(part)
return blocks
return str(content)
def _convert_image(part):
"""将 OpenAI image_url 格式转为 Anthropic image 格式"""
url_data = part.get('image_url', {})
url = url_data.get('url', '') if isinstance(url_data, dict) else str(url_data)
if url.startswith('data:'):
media_type, _, b64 = url.partition(';base64,')
return {
'type': 'image',
'source': {
'type': 'base64',
'media_type': media_type.replace('data:', '') or 'image/png',
'data': b64,
},
}
return {'type': 'image', 'source': {'type': 'url', 'url': url}}
def _convert_tools(tools):
"""将 OpenAI tools 转为 Anthropic tools 格式（兼容 Cursor 扁平格式）"""
result = []
for tool in tools:
if tool.get('type') == 'function' and 'function' in tool:
func = tool['function']
result.append({
'name': func.get('name', ''),
'description': func.get('description', ''),
'input_schema': func.get('parameters', {'type': 'object', 'properties': {}}),
})
elif 'name' in tool and 'input_schema' in tool:
result.append({
'name': tool.get('name', ''),
'description': tool.get('description', ''),
'input_schema': tool.get('input_schema', {'type': 'object', 'properties': {}}),
})
return result
def _to_blocks(content):
"""将 content 统一转为 blocks 列表"""
if isinstance(content, str):
return [{'type': 'text', 'text': content}] if content else []
if isinstance(content, list):
return list(content)
return [{'type': 'text', 'text': str(content)}] if content else []
def _merge_same_role(messages):
"""合并相邻同角色消息（Anthropic 要求角色必须交替）"""
if not messages:
return messages
merged = [messages[0]]
for msg in messages[1:]:
if msg['role'] == merged[-1]['role']:
prev = _to_blocks(merged[-1]['content'])
curr = _to_blocks(msg['content'])
merged[-1]['content'] = prev + curr
else:
merged.append(msg)
return merged

405
adapters/openai_compat_fixer.py Normal file
View file

@ -0,0 +1,405 @@
"""OpenAI 格式修复
这个模块专门处理 OpenAI Chat Completions 兼容层里的“脏活”：
- 请求方向：把 Cursor 发来的近似 OpenAI 格式修整成更标准的请求
- 响应方向：把上游返回的近似 OpenAI 格式修整成 Cursor 更容易消费的结果
这里之所以集中做兼容性修复，而不是散落在路由层，是因为这些规则本质上属于
“协议清洗”而不是“请求编排”。路由层只应该关心把请求送到哪里，修复规则则应该
在适配层统一收口，避免两条主链路各自维护一份类似逻辑。
"""
from __future__ import annotations
import json
import logging
from typing import Any
from utils.http import gen_id
from utils.think_tag import extract_from_text
from utils.tool_fixer import normalize_args, repair_str_replace_args
logger = logging.getLogger(__name__)
JsonDict = dict[str, Any]
# ─── 请求预处理 ───────────────────────────────────
def normalize_request(payload: JsonDict, upstream_model: str | None = None) -> JsonDict:
"""预处理 Cursor 发来的 OpenAI 风格请求。
这个函数只做“让请求更像标准 OpenAI CC”的整理，不负责路由或网络层决策。
当前处理的重点有两类：
1. Cursor 偶尔会在 CC 端点混入 Anthropic 风格内容块，需要先转回 OpenAI 语义。
2. 工具定义和 tool_choice 可能是 Cursor 的便捷写法，需要标准化后再发给上游。
"""
if upstream_model:
payload['model'] = upstream_model
if 'messages' in payload:
payload['messages'] = _convert_anthropic_messages(payload['messages'])
if 'tools' not in payload:
return payload
payload['tools'] = [_normalize_tool_definition(tool) for tool in payload['tools']]
_normalize_tool_choice(payload)
return payload
# ─── 消息兼容转换 ─────────────────────────────────
def _convert_anthropic_messages(messages: Any) -> Any:
"""将消息中的 Anthropic tool_use/tool_result 块转回 OpenAI 风格消息。
Cursor 在少数场景下会把 Anthropic 风格内容块直接发到
`/v1/chat/completions`。如果不在这里先转换，后续上游即使是 OpenAI 兼容接口，
也未必能理解这类内容块。
"""
if not isinstance(messages, list):
return messages
converted: list[JsonDict] = []
for message in messages:
converted.extend(_convert_single_message(message))
return converted
def _convert_single_message(message: Any) -> list[JsonDict]:
"""将单条消息转换为 1 条或多条 OpenAI 风格消息。"""
if not isinstance(message, dict):
return [message]
content = message.get('content')
if not isinstance(content, list):
return [message]
has_tool_use, has_tool_result = _detect_tool_blocks(content)
if not has_tool_use and not has_tool_result:
return [message]
role = message.get('role', '')
if role == 'assistant' and has_tool_use:
return [_convert_assistant_tool_use_message(content)]
if has_tool_result:
return _convert_tool_result_message(role, content)
return [message]
def _detect_tool_blocks(content: list[Any]) -> tuple[bool, bool]:
"""识别内容块里是否包含 Anthropic 风格工具调用或工具结果。"""
has_tool_use = any(
isinstance(block, dict) and block.get('type') == 'tool_use'
for block in content
)
has_tool_result = any(
isinstance(block, dict) and block.get('type') == 'tool_result'
for block in content
)
return has_tool_use, has_tool_result
def _convert_assistant_tool_use_message(content: list[Any]) -> JsonDict:
"""将 assistant 的 tool_use 内容块转为 OpenAI tool_calls。"""
text_parts: list[str] = []
tool_calls: list[JsonDict] = []
for block in content:
if not isinstance(block, dict):
continue
if block.get('type') == 'text':
text_parts.append(block.get('text', ''))
elif block.get('type') == 'tool_use':
tool_calls.append({
'id': block.get('id', gen_id('call_')),
'type': 'function',
'function': {
'name': block.get('name', ''),
'arguments': json.dumps(block.get('input', {}), ensure_ascii=False),
},
})
result: JsonDict = {
'role': 'assistant',
'content': '\n'.join(text_parts) if text_parts else None,
}
if tool_calls:
result['tool_calls'] = tool_calls
return result
def _convert_tool_result_message(role: str, content: list[Any]) -> list[JsonDict]:
"""将 tool_result 块拆成 OpenAI 的 tool 消息，并保留其余内容块。"""
converted: list[JsonDict] = []
other_parts: list[Any] = []
for block in content:
if not isinstance(block, dict):
continue
if block.get('type') == 'tool_result':
converted.append({
'role': 'tool',
'tool_call_id': block.get('tool_use_id', ''),
'content': _stringify_tool_result_content(block.get('content', '')),
})
else:
other_parts.append(block)
if other_parts:
converted.append({'role': role, 'content': other_parts})
return converted
def _stringify_tool_result_content(content: Any) -> str:
"""将 tool_result 的 content 规范为字符串。
OpenAI 的 tool 消息内容天然更偏向字符串；而 Anthropic 的 tool_result 允许列表块。
这里做一次降维，避免后续上游把结构化结果误当成普通消息块。
"""
if isinstance(content, str):
return content
if isinstance(content, list):
return '\n'.join(
block.get('text', '')
for block in content
if isinstance(block, dict) and block.get('type') == 'text'
)
return str(content)
def _normalize_tool_definition(tool: Any) -> Any:
"""将 Cursor 可能使用的扁平工具定义补成标准 OpenAI function tool。
这里不主动过滤未知字段，只做最小标准化，避免在兼容层里过早丢失调用方提供的
额外上下文。
"""
if not isinstance(tool, dict):
return tool
if tool.get('type') == 'function' and 'function' in tool:
return tool
if 'name' not in tool:
return tool
return {
'type': 'function',
'function': {
'name': tool.get('name', ''),
'description': tool.get('description', ''),
'parameters': (
tool.get('input_schema')
or tool.get('parameters')
or {'type': 'object', 'properties': {}}
),
},
}
def _normalize_tool_choice(payload: JsonDict) -> None:
"""规范化 tool_choice。
这里保留当前项目已有的映射约定：
- `{"type": "auto"}` → `"auto"`
- `{"type": "any"}` → `"required"`
这样做是因为部分上游只接受 OpenAI 常见的字符串写法，而不接受 Cursor/Anthropic
风格的对象写法。
"""
tool_choice = payload.get('tool_choice')
if not isinstance(tool_choice, dict):
return
if tool_choice.get('type') == 'auto':
payload['tool_choice'] = 'auto'
elif tool_choice.get('type') == 'any':
payload['tool_choice'] = 'required'
# ─── 非流式响应修复 ───────────────────────────────
def fix_response(data: Any) -> Any:
"""修复上游返回的非流式 OpenAI 响应。"""
if not isinstance(data, dict):
return data
for choice in data.get('choices') or []:
_fix_response_choice(choice)
return data
def _fix_response_choice(choice: Any) -> None:
"""修复单个非流式 choice。"""
if not isinstance(choice, dict):
return
message = choice.get('message') or {}
if not isinstance(message, dict):
return
_promote_reasoning_field(message)
_extract_reasoning_from_content(message)
_convert_legacy_message_function_call(message, choice)
_fix_tool_calls(message, choice)
def _promote_reasoning_field(container: JsonDict) -> None:
"""兼容不同上游返回的 reasoning 字段命名差异。"""
if 'reasoningContent' in container and 'reasoning_content' not in container:
container['reasoning_content'] = container.pop('reasoningContent')
def _extract_reasoning_from_content(message: JsonDict) -> None:
"""从 `<think>...</think>` 中提取 reasoning_content。
有些上游把思考内容直接塞进 content 字符串里，而不是单独返回 reasoning 字段。
这里主动提取，是为了让 Cursor 端更稳定地展示思考过程。
"""
content = message.get('content') or ''
if not isinstance(content, str):
return
if '<think>' not in content or message.get('reasoning_content'):
return
cleaned, reasoning = extract_from_text(content)
if not reasoning:
return
message['reasoning_content'] = reasoning
message['content'] = cleaned
logger.info('已提取 <think> 标签内容并映射为 reasoning_content，长度=%s', len(reasoning))
def _convert_legacy_message_function_call(message: JsonDict, choice: JsonDict) -> None:
"""将旧版 function_call 字段升级为新版 tool_calls。"""
if 'function_call' not in message or 'tool_calls' in message:
return
function_call = message.pop('function_call') or {}
message['tool_calls'] = [{
'id': gen_id('call_'),
'type': 'function',
'function': {
'name': function_call.get('name', ''),
'arguments': function_call.get('arguments', '{}'),
},
}]
_rewrite_function_call_finish_reason(choice)
# ─── 流式 chunk 修复 ──────────────────────────────
def fix_stream_chunk(data: Any) -> Any:
"""修复上游返回的流式 OpenAI chunk。"""
if not isinstance(data, dict):
return data
for choice in data.get('choices') or []:
_fix_stream_choice(choice)
return data
def _fix_stream_choice(choice: Any) -> None:
"""修复单个流式 choice。"""
if not isinstance(choice, dict):
return
delta = choice.get('delta') or {}
if not isinstance(delta, dict):
return
_promote_reasoning_field(delta)
_convert_legacy_delta_function_call(delta, choice)
_ensure_stream_tool_calls(delta)
_rewrite_function_call_finish_reason(choice)
def _convert_legacy_delta_function_call(delta: JsonDict, choice: JsonDict) -> None:
"""将流式旧版 function_call 增量升级为 tool_calls 增量。"""
if 'function_call' not in delta or 'tool_calls' in delta:
return
function_call = delta.pop('function_call') or {}
tool_call: JsonDict = {'index': 0, 'type': 'function', 'function': {}}
if 'name' in function_call:
tool_call['id'] = gen_id('call_')
tool_call['function']['name'] = function_call['name']
if 'arguments' in function_call:
tool_call['function']['arguments'] = function_call['arguments']
delta['tool_calls'] = [tool_call]
_rewrite_function_call_finish_reason(choice)
def _ensure_stream_tool_calls(delta: JsonDict) -> None:
"""补全流式 tool_calls 的最小必需字段。
流式增量中的 tool_calls 往往是不完整片段，这里只补齐索引、ID、类型等元信息，
不主动改写 arguments 内容，避免破坏增量拼接语义。
"""
for tool_call in delta.get('tool_calls') or []:
if 'index' not in tool_call:
tool_call['index'] = 0
function_data = tool_call.get('function') or {}
if 'id' in tool_call or 'name' in function_data:
if not tool_call.get('id'):
tool_call['id'] = gen_id('call_')
if 'type' not in tool_call:
tool_call['type'] = 'function'
# ─── tool_calls 修复 ──────────────────────────────
def _fix_tool_calls(message: JsonDict, choice: JsonDict) -> None:
"""修复非流式消息中的 tool_calls 字段。"""
tool_calls = message.get('tool_calls')
if not tool_calls:
return
for index, tool_call in enumerate(tool_calls):
_fill_tool_call_metadata(tool_call, index=index)
_normalize_tool_call_arguments(tool_call)
if choice.get('finish_reason') not in ('tool_calls', 'function_call'):
choice['finish_reason'] = 'tool_calls'
def _fill_tool_call_metadata(tool_call: JsonDict, *, index: int) -> None:
"""补齐非流式 tool_call 的通用元数据。"""
if not tool_call.get('id'):
tool_call['id'] = gen_id('call_')
if 'index' not in tool_call:
tool_call['index'] = index
if tool_call.get('type') != 'function':
tool_call['type'] = 'function'
def _normalize_tool_call_arguments(tool_call: JsonDict) -> None:
"""规范化 tool_call 参数。
这里会顺带调用工具参数修复器，原因是很多兼容性问题不在协议层，而在工具参数本身：
比如 `file_path`/`path` 命名差异、智能引号、StrReplace 精确匹配失败等。
"""
function_data = tool_call.get('function') or {}
raw_arguments = function_data.get('arguments', '{}')
try:
arguments = (
json.loads(raw_arguments)
if isinstance(raw_arguments, str)
else (raw_arguments or {})
)
except json.JSONDecodeError:
arguments = {}
arguments = normalize_args(arguments)
arguments = repair_str_replace_args(function_data.get('name', ''), arguments)
function_data['arguments'] = json.dumps(arguments, ensure_ascii=False)
def _rewrite_function_call_finish_reason(choice: JsonDict) -> None:
"""将旧版 finish_reason=function_call 升级为 tool_calls。"""
if choice.get('finish_reason') == 'function_call':
choice['finish_reason'] = 'tool_calls'

267
adapters/openai_fixer.py
View file

@ -1,267 +0,0 @@
"""OpenAI 格式修复
修复 Cursor 发出的 OpenAI 格式请求和上游返回的响应中的各种兼容性问题：
请求修复: Cursor 扁平格式 tools → 标准嵌套格式, tool_choice 规范化
响应修复: reasoningContent → reasoning_content, <think> 标签提取,
function_call → tool_calls, tool_calls 字段补全, 参数修复
"""
import json
import logging
from utils.http import gen_id
from utils.tool_fixer import normalize_args, repair_str_replace_args
from utils.think_tag import extract_from_text
logger = logging.getLogger(__name__)
# ─── 请求预处理 ───────────────────────────────────
def normalize_request(payload, upstream_model=None):
"""预处理 Cursor 发来的 OpenAI 格式请求"""
if upstream_model:
payload['model'] = upstream_model
# Cursor 可能在 CC 端点发送 Anthropic 格式的 tool_use/tool_result 消息
if 'messages' in payload:
payload['messages'] = _convert_anthropic_messages(payload['messages'])
if 'tools' not in payload:
return payload
# 修复 Cursor 可能发出的扁平格式 tools
normalized = []
for tool in payload['tools']:
if tool.get('type') == 'function' and 'function' in tool:
normalized.append(tool)
elif 'name' in tool:
normalized.append({
'type': 'function',
'function': {
'name': tool.get('name', ''),
'description': tool.get('description', ''),
'parameters': tool.get('input_schema')
or tool.get('parameters')
or {'type': 'object', 'properties': {}},
},
})
else:
normalized.append(tool)
payload['tools'] = normalized
# tool_choice 规范化
tc = payload.get('tool_choice')
if isinstance(tc, dict):
if tc.get('type') == 'auto':
payload['tool_choice'] = 'auto'
elif tc.get('type') == 'any':
payload['tool_choice'] = 'required'
return payload
def _convert_anthropic_messages(messages):
"""将消息中的 Anthropic 格式 tool_use/tool_result 转为 OpenAI 格式
Cursor 有时在 CC 端点中发送 Anthropic 风格的内容块：
assistant: [{"type":"tool_use", "id":"...", "name":"Read", "input":{...}}]
user: [{"type":"tool_result", "tool_use_id":"...", "content":[...]}]
OpenAI 格式应为：
assistant: {"tool_calls":[{"id":"...", "function":{"name":"Read","arguments":"..."}}]}
tool: {"tool_call_id":"...", "content":"..."}
"""
converted = []
for msg in messages:
content = msg.get('content')
if not isinstance(content, list):
converted.append(msg)
continue
has_tool_use = any(
isinstance(b, dict) and b.get('type') == 'tool_use' for b in content
)
has_tool_result = any(
isinstance(b, dict) and b.get('type') == 'tool_result' for b in content
)
if not has_tool_use and not has_tool_result:
converted.append(msg)
continue
role = msg.get('role', '')
if role == 'assistant' and has_tool_use:
text_parts = []
tool_calls = []
for block in content:
if not isinstance(block, dict):
continue
if block.get('type') == 'text':
text_parts.append(block.get('text', ''))
elif block.get('type') == 'tool_use':
tool_calls.append({
'id': block.get('id', gen_id('call_')),
'type': 'function',
'function': {
'name': block.get('name', ''),
'arguments': json.dumps(
block.get('input', {}), ensure_ascii=False
),
},
})
new_msg = {'role': 'assistant'}
new_msg['content'] = '\n'.join(text_parts) if text_parts else None
if tool_calls:
new_msg['tool_calls'] = tool_calls
converted.append(new_msg)
elif has_tool_result:
other_parts = []
for block in content:
if not isinstance(block, dict):
continue
if block.get('type') == 'tool_result':
rc = block.get('content', '')
if isinstance(rc, list):
rc = '\n'.join(
b.get('text', '') for b in rc
if isinstance(b, dict) and b.get('type') == 'text'
)
elif not isinstance(rc, str):
rc = str(rc)
converted.append({
'role': 'tool',
'tool_call_id': block.get('tool_use_id', ''),
'content': rc,
})
else:
other_parts.append(block)
if other_parts:
converted.append({'role': role, 'content': other_parts})
else:
converted.append(msg)
return converted
# ─── 非流式响应修复 ───────────────────────────────
def fix_response(data):
"""修复上游返回的非流式 OpenAI 响应"""
if not isinstance(data, dict):
return data
for choice in (data.get('choices') or []):
msg = choice.get('message') or {}
# reasoningContent → reasoning_content
if 'reasoningContent' in msg and 'reasoning_content' not in msg:
msg['reasoning_content'] = msg.pop('reasoningContent')
# <think> 标签 → reasoning_content
content = msg.get('content') or ''
if isinstance(content, str) and '<think>' in content and not msg.get('reasoning_content'):
cleaned, reasoning = extract_from_text(content)
if reasoning:
msg['reasoning_content'] = reasoning
msg['content'] = cleaned
logger.info(f'提取 <think> 标签 → reasoning_content ({len(reasoning)} 字符)')
# 旧版 function_call → 新版 tool_calls
if 'function_call' in msg and 'tool_calls' not in msg:
fc = msg.pop('function_call')
msg['tool_calls'] = [{
'id': gen_id('call_'),
'type': 'function',
'function': {
'name': fc.get('name', ''),
'arguments': fc.get('arguments', '{}'),
},
}]
if choice.get('finish_reason') == 'function_call':
choice['finish_reason'] = 'tool_calls'
# 修复 tool_calls 字段
_fix_tool_calls(msg, choice)
return data
# ─── 流式 chunk 修复 ──────────────────────────────
def fix_stream_chunk(data):
"""修复上游返回的流式 OpenAI chunk"""
if not isinstance(data, dict):
return data
for choice in (data.get('choices') or []):
delta = choice.get('delta') or {}
# reasoningContent → reasoning_content
if 'reasoningContent' in delta and 'reasoning_content' not in delta:
delta['reasoning_content'] = delta.pop('reasoningContent')
# 旧版 function_call → tool_calls
if 'function_call' in delta and 'tool_calls' not in delta:
fc = delta.pop('function_call')
tc = {'index': 0, 'type': 'function', 'function': {}}
if 'name' in fc:
tc['id'] = gen_id('call_')
tc['function']['name'] = fc['name']
if 'arguments' in fc:
tc['function']['arguments'] = fc['arguments']
delta['tool_calls'] = [tc]
if choice.get('finish_reason') == 'function_call':
choice['finish_reason'] = 'tool_calls'
# 补全 tool_calls 字段
for tc in (delta.get('tool_calls') or []):
if 'index' not in tc:
tc['index'] = 0
func = tc.get('function') or {}
if 'id' in tc or 'name' in func:
if not tc.get('id'):
tc['id'] = gen_id('call_')
if 'type' not in tc:
tc['type'] = 'function'
if choice.get('finish_reason') == 'function_call':
choice['finish_reason'] = 'tool_calls'
return data
# ─── 内部辅助 ─────────────────────────────────────
def _fix_tool_calls(msg, choice):
"""修复消息中的 tool_calls 字段"""
tool_calls = msg.get('tool_calls')
if not tool_calls:
return
for i, tc in enumerate(tool_calls):
if not tc.get('id'):
tc['id'] = gen_id('call_')
if 'index' not in tc:
tc['index'] = i
if tc.get('type') != 'function':
tc['type'] = 'function'
func = tc.get('function', {})
args_raw = func.get('arguments', '{}')
try:
args = json.loads(args_raw) if isinstance(args_raw, str) else (args_raw or {})
except json.JSONDecodeError:
args = {}
args = normalize_args(args)
args = repair_str_replace_args(func.get('name', ''), args)
func['arguments'] = json.dumps(args, ensure_ascii=False)
if choice.get('finish_reason') not in ('tool_calls', 'function_call'):
choice['finish_reason'] = 'tool_calls'

533
adapters/responses_adapter.py
View file

@ -1,533 +0,0 @@
"""Responses API 适配
Cursor 对 GPT/Claude-Opus 等模型使用 /v1/responses 格式。
本模块将 Responses 格式与 Chat Completions 格式互相转换：
请求: Responses → CC
响应: CC → Responses（非流式 + 流式）
流式: 支持从 CC chunks 或 Anthropic SSE 事件直接转换
"""
import json
import logging
from utils.http import gen_id
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════
# 请求转换: Responses → CC
# ═══════════════════════════════════════════════════════════
def responses_to_cc(payload):
"""将 /v1/responses 请求转换为 /v1/chat/completions 格式"""
messages = []
if payload.get('instructions'):
messages.append({'role': 'system', 'content': payload['instructions']})
input_data = payload.get('input', [])
if isinstance(input_data, str):
messages.append({'role': 'user', 'content': input_data})
elif isinstance(input_data, list):
_convert_input_items(input_data, messages)
result = {
'model': payload.get('model', ''),
'messages': messages,
'stream': payload.get('stream', False),
}
if 'tools' in payload:
result['tools'] = _convert_tools(payload['tools'])
for key in ('temperature', 'top_p'):
if key in payload:
result[key] = payload[key]
if 'max_output_tokens' in payload:
result['max_tokens'] = payload['max_output_tokens']
if 'tool_choice' in payload:
result['tool_choice'] = payload['tool_choice']
return result
# ═══════════════════════════════════════════════════════════
# 非流式响应转换: CC → Responses
# ═══════════════════════════════════════════════════════════
def cc_to_responses(cc_resp, model=''):
"""将 CC 响应转换为 Responses 格式"""
choice = (cc_resp.get('choices') or [{}])[0]
msg = choice.get('message') or {}
finish = choice.get('finish_reason', 'stop')
output = []
if msg.get('reasoning_content'):
output.append({
'type': 'reasoning',
'id': gen_id('rs_'),
'summary': [{'type': 'summary_text', 'text': msg['reasoning_content']}],
})
if msg.get('content'):
output.append({
'type': 'message',
'id': gen_id('msg_'),
'status': 'completed',
'role': 'assistant',
'content': [{'type': 'output_text', 'text': msg['content']}],
})
for tc in (msg.get('tool_calls') or []):
func = tc.get('function') or {}
output.append({
'type': 'function_call',
'id': gen_id('fc_'),
'status': 'completed',
'call_id': tc.get('id', gen_id('call_')),
'name': func.get('name', ''),
'arguments': func.get('arguments', '{}'),
})
usage = cc_resp.get('usage', {})
return {
'id': cc_resp.get('id', gen_id('resp_')),
'object': 'response',
'status': 'incomplete' if finish == 'length' else 'completed',
'model': model or cc_resp.get('model', ''),
'output': output,
'usage': {
'input_tokens': usage.get('prompt_tokens', 0),
'output_tokens': usage.get('completion_tokens', 0),
'total_tokens': usage.get('total_tokens', 0),
},
}
# ═══════════════════════════════════════════════════════════
# 流式转换器: CC chunks / Anthropic SSE → Responses SSE
# ═══════════════════════════════════════════════════════════
class ResponsesStreamConverter:
"""有状态转换器：将 CC 流式 chunk 或 Anthropic SSE 事件转为 Responses SSE 事件"""
def __init__(self, response_id=None, model=''):
self.resp_id = response_id or gen_id('resp_')
self.model = model
# 思考内容缓冲
self._rs_buf = ''
self._rs_started = False
self._rs_closed = False
self._rs_id = gen_id('rs_')
# 文本内容缓冲
self._text_buf = ''
self._text_started = False
self._text_closed = False
self._msg_id = gen_id('msg_')
# 工具调用缓冲 {index: {name, args, call_id, fc_id}}
self._tools = {}
self._output_items = []
self._finished = False
self._input_tokens = 0
# ─── 公开接口 ─────────────────────────────────
def start_events(self):
"""生成流开始事件"""
return [self._sse('response.created', {
'id': self.resp_id, 'object': 'response',
'status': 'in_progress', 'model': self.model, 'output': [],
})]
def process_cc_chunk(self, chunk):
"""处理 CC 格式的流式 chunk，返回 Responses SSE 事件列表"""
events = []
for choice in (chunk.get('choices') or []):
delta = choice.get('delta') or {}
finish = choice.get('finish_reason')
if delta.get('reasoning_content'):
events.extend(self._on_reasoning(delta['reasoning_content']))
if delta.get('content') is not None and delta['content'] != '':
events.extend(self._on_text(delta['content']))
for tc in (delta.get('tool_calls') or []):
events.extend(self._on_tool_call(tc))
if finish and not self._finished:
self._finished = True
events.extend(self._do_finish(finish, chunk.get('usage')))
return events
def process_anthropic_event(self, event_type, event_data):
"""直接处理 Anthropic SSE 事件（跳过 CC 中间转换，更高效）"""
events = []
if event_type == 'message_start':
usage = event_data.get('message', {}).get('usage', {})
self._input_tokens = usage.get('input_tokens', 0)
elif event_type == 'content_block_start':
block = event_data.get('content_block', {})
btype = block.get('type', '')
if btype == 'thinking' and not self._rs_started:
self._rs_started = True
events.append(self._sse('response.output_item.added', {
'type': 'reasoning', 'id': self._rs_id, 'summary': [],
}))
elif btype == 'text':
events.extend(self._ensure_text_started())
elif btype == 'tool_use':
events.extend(self._start_tool_from_block(block))
elif event_type == 'content_block_delta':
delta = event_data.get('delta', {})
dtype = delta.get('type', '')
if dtype == 'thinking_delta' and delta.get('thinking'):
self._rs_buf += delta['thinking']
events.append(self._sse('response.reasoning_summary_text.delta', {
'type': 'summary_text', 'delta': delta['thinking'],
}))
elif dtype == 'text_delta' and delta.get('text'):
self._text_buf += delta['text']
events.append(self._sse('response.output_text.delta', {
'type': 'output_text', 'delta': delta['text'],
}))
elif dtype == 'input_json_delta' and delta.get('partial_json') and self._tools:
idx = max(self._tools.keys())
self._tools[idx]['args'] += delta['partial_json']
events.append(self._sse('response.function_call_arguments.delta', {
'type': 'function_call', 'delta': delta['partial_json'],
}))
elif event_type == 'message_delta':
delta = event_data.get('delta', {})
stop = delta.get('stop_reason', 'end_turn')
usage = event_data.get('usage', {})
finish = {'tool_use': 'tool_calls', 'max_tokens': 'length'}.get(stop, 'stop')
if not self._finished:
self._finished = True
u = {
'input_tokens': self._input_tokens,
'output_tokens': usage.get('output_tokens', 0),
'total_tokens': self._input_tokens + usage.get('output_tokens', 0),
}
events.extend(self._do_finish(finish, u))
return events
def finalize(self):
"""流结束时补发未关闭的事件"""
if self._finished:
return []
self._finished = True
return self._do_finish('stop', None)
# ─── 内部事件处理 ─────────────────────────────
def _on_reasoning(self, text):
"""处理思考内容 delta"""
events = []
if not self._rs_started:
self._rs_started = True
events.append(self._sse('response.output_item.added', {
'type': 'reasoning', 'id': self._rs_id, 'summary': [],
}))
self._rs_buf += text
events.append(self._sse('response.reasoning_summary_text.delta', {
'type': 'summary_text', 'delta': text,
}))
return events
def _on_text(self, text):
"""处理文本内容 delta"""
events = self._ensure_text_started()
self._text_buf += text
events.append(self._sse('response.output_text.delta', {
'type': 'output_text', 'delta': text,
}))
return events
def _on_tool_call(self, tc):
"""处理工具调用 delta"""
events = []
idx = tc.get('index', 0)
func = tc.get('function') or {}
if idx not in self._tools:
if self._rs_started and not self._rs_closed:
events.extend(self._close_reasoning())
if self._text_started and not self._text_closed:
events.extend(self._close_text())
call_id = tc.get('id', gen_id('call_'))
name = func.get('name', '')
fc_id = gen_id('fc_')
self._tools[idx] = {'name': name, 'args': '', 'call_id': call_id, 'fc_id': fc_id}
events.append(self._sse('response.output_item.added', {
'type': 'function_call', 'id': fc_id,
'status': 'in_progress', 'call_id': call_id,
'name': name, 'arguments': '',
}))
if func.get('name'):
self._tools[idx]['name'] = func['name']
if func.get('arguments', ''):
self._tools[idx]['args'] += func['arguments']
events.append(self._sse('response.function_call_arguments.delta', {
'type': 'function_call', 'delta': func['arguments'],
}))
return events
def _ensure_text_started(self):
"""确保文本输出项已开始"""
events = []
if self._rs_started and not self._rs_closed:
events.extend(self._close_reasoning())
if not self._text_started:
self._text_started = True
events.append(self._sse('response.output_item.added', {
'type': 'message', 'id': self._msg_id,
'status': 'in_progress', 'role': 'assistant', 'content': [],
}))
events.append(self._sse('response.content_part.added', {
'type': 'output_text', 'text': '',
}))
return events
def _start_tool_from_block(self, block):
"""从 Anthropic tool_use block 开始新的工具调用"""
events = []
if self._rs_started and not self._rs_closed:
events.extend(self._close_reasoning())
if self._text_started and not self._text_closed:
events.extend(self._close_text())
idx = len(self._tools)
tool_id = block.get('id', gen_id('toolu_'))
name = block.get('name', '')
fc_id = gen_id('fc_')
self._tools[idx] = {'name': name, 'args': '', 'call_id': tool_id, 'fc_id': fc_id}
events.append(self._sse('response.output_item.added', {
'type': 'function_call', 'id': fc_id,
'status': 'in_progress', 'call_id': tool_id,
'name': name, 'arguments': '',
}))
return events
# ─── 关闭/结束事件 ────────────────────────────
def _close_reasoning(self):
if self._rs_closed:
return []
self._rs_closed = True
rs = {
'type': 'reasoning', 'id': self._rs_id,
'summary': [{'type': 'summary_text', 'text': self._rs_buf}],
}
self._output_items.append(rs)
return [
self._sse('response.reasoning_summary_text.done', {
'type': 'summary_text', 'text': self._rs_buf,
}),
self._sse('response.output_item.done', rs),
]
def _close_text(self):
if self._text_closed:
return []
self._text_closed = True
msg = {
'type': 'message', 'id': self._msg_id,
'status': 'completed', 'role': 'assistant',
'content': [{'type': 'output_text', 'text': self._text_buf}],
}
self._output_items.append(msg)
return [
self._sse('response.output_text.done', {'type': 'output_text', 'text': self._text_buf}),
self._sse('response.output_item.done', msg),
]
def _do_finish(self, finish_reason, usage):
"""生成流结束的所有关闭事件"""
events = []
if self._rs_started and not self._rs_closed:
events.extend(self._close_reasoning())
if self._text_started and not self._text_closed:
events.extend(self._close_text())
for idx in sorted(self._tools.keys()):
buf = self._tools[idx]
events.append(self._sse('response.function_call_arguments.done', {
'type': 'function_call', 'arguments': buf['args'],
}))
fc = {
'type': 'function_call', 'id': buf['fc_id'],
'status': 'completed', 'call_id': buf['call_id'],
'name': buf['name'], 'arguments': buf['args'],
}
events.append(self._sse('response.output_item.done', fc))
self._output_items.append(fc)
usage_data = usage if isinstance(usage, dict) else {}
events.append(self._sse('response.completed', {
'id': self.resp_id, 'object': 'response',
'status': 'incomplete' if finish_reason == 'length' else 'completed',
'model': self.model, 'output': self._output_items, 'usage': usage_data,
}))
return events
def _sse(self, event_type, data):
"""构建 SSE 事件字符串"""
return f'event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n'
# ═══════════════════════════════════════════════════════════
# 内部辅助函数
# ═══════════════════════════════════════════════════════════
def _convert_input_items(items, messages):
"""将 Responses input 数组转换为 CC messages"""
i = 0
while i < len(items):
item = items[i]
if isinstance(item, str):
messages.append({'role': 'user', 'content': item})
i += 1
continue
if not isinstance(item, dict):
i += 1
continue
item_type = item.get('type', '')
role = item.get('role', '')
# 简单角色消息（无 type 字段）
if role and not item_type:
content = item.get('content', '')
if isinstance(content, list):
content = _extract_text(content)
messages.append({'role': role, 'content': content or ''})
i += 1
continue
# Responses message 对象
if item_type == 'message' or (role and not item_type):
role = item.get('role', 'assistant')
content = _extract_text(item.get('content', []))
msg = {'role': role, 'content': content or ''}
if role == 'assistant':
tool_calls, consumed = _collect_function_calls(items, i + 1)
if tool_calls:
msg['tool_calls'] = tool_calls
if not msg['content']:
msg['content'] = None
messages.append(msg)
i += 1 + consumed
continue
messages.append(msg)
i += 1
continue
# function_call（工具调用）
if item_type == 'function_call':
tc = {
'id': item.get('call_id') or gen_id('call_'),
'type': 'function',
'function': {
'name': item.get('name', ''),
'arguments': item.get('arguments', '{}'),
},
}
if messages and messages[-1]['role'] == 'assistant':
messages[-1].setdefault('tool_calls', []).append(tc)
if not messages[-1].get('content'):
messages[-1]['content'] = None
else:
messages.append({'role': 'assistant', 'content': None, 'tool_calls': [tc]})
i += 1
continue
# function_call_output（工具结果）
if item_type == 'function_call_output':
output = item.get('output', '')
if not isinstance(output, str):
output = json.dumps(output, ensure_ascii=False)
messages.append({
'role': 'tool',
'tool_call_id': item.get('call_id', ''),
'content': output,
})
i += 1
continue
if role:
messages.append({'role': role, 'content': str(item.get('content', ''))})
i += 1
def _collect_function_calls(items, start):
"""收集紧随 assistant message 之后的连续 function_call 项"""
tool_calls = []
j = start
while j < len(items):
nxt = items[j]
if isinstance(nxt, dict) and nxt.get('type') == 'function_call':
tool_calls.append({
'id': nxt.get('call_id') or gen_id('call_'),
'type': 'function',
'function': {
'name': nxt.get('name', ''),
'arguments': nxt.get('arguments', '{}'),
},
})
j += 1
else:
break
return tool_calls, j - start
def _extract_text(content):
"""从 content 中提取纯文本"""
if isinstance(content, str):
return content
if not isinstance(content, list):
return str(content) if content else ''
texts = []
for part in content:
if isinstance(part, str):
texts.append(part)
elif isinstance(part, dict):
t = part.get('type', '')
if t in ('output_text', 'input_text', 'text'):
texts.append(part.get('text', ''))
elif t == 'refusal':
texts.append(part.get('refusal', ''))
return '\n'.join(texts) if texts else ''
def _convert_tools(tools):
"""将 Responses tools 转为 CC tools 格式"""
result = []
for t in tools:
if t.get('type') != 'function':
continue
if 'function' in t:
result.append(t)
else:
result.append({
'type': 'function',
'function': {
'name': t.get('name', ''),
'description': t.get('description', ''),
'parameters': t.get('parameters', {'type': 'object', 'properties': {}}),
},
})
return result

1052
adapters/responses_cc_adapter.py Normal file
View file

File diff suppressed because it is too large Load diff

436
routes/chat.py
View file

@ -1,216 +1,388 @@
"""路由: /v1/chat/completions
处理 Cursor 发来的 OpenAI Chat Completions 格式请求。
根据模型映射的 backend 字段分发到 OpenAI 或 Anthropic 后端。
根据模型映射的后端类型，转发到 OpenAI 兼容接口、Anthropic Messages 接口，
或原生 OpenAI Responses 接口。
"""
from __future__ import annotations
import json
import logging
from typing import Any
from flask import Blueprint, request, jsonify
from flask import Blueprint, jsonify, request
import settings
from config import Config
from adapters.openai_fixer import normalize_request, fix_response, fix_stream_chunk
from adapters.openai_anthropic import (
cc_to_messages_request, messages_to_cc_response, AnthropicStreamConverter,
from adapters.cc_anthropic_adapter import (
AnthropicStreamConverter,
cc_to_messages_request,
messages_to_cc_response,
)
from adapters.openai_compat_fixer import fix_response, fix_stream_chunk, normalize_request
from adapters.responses_cc_adapter import (
ResponsesToCCStreamConverter,
cc_to_responses_request,
responses_to_cc,
responses_to_cc_response,
)
from config import Config
from routes.common import (
RouteContext,
build_anthropic_target,
build_openai_target,
build_responses_target,
build_route_context,
chat_error_chunk,
log_route_context,
log_usage,
sse_data_message,
)
from adapters.responses_adapter import responses_to_cc
from utils.http import (
build_openai_headers, build_anthropic_headers,
forward_request, sse_response,
iter_openai_sse, iter_anthropic_sse,
forward_request,
iter_anthropic_sse,
iter_openai_sse,
iter_responses_sse,
sse_response,
)
from utils.think_tag import ThinkTagExtractor
logger = logging.getLogger(__name__)
def _dbg(msg):
"""DEBUG 模式下输出详细日志"""
if Config.DEBUG:
logger.info(f'[调试] {msg}')
bp = Blueprint('chat', __name__)
def _dbg(message: str) -> None:
"""仅在调试模式下输出详细日志。"""
if Config.DEBUG:
logger.info('[聊天补全调试] %s', message)
@bp.route('/v1/chat/completions', methods=['POST'])
def chat_completions():
"""处理聊天补全请求并按模型映射分发到不同后端。"""
payload = request.get_json(force=True)
payload, message_count = _normalize_chat_payload(payload)
client_model = payload.get('model', 'unknown')
is_stream = payload.get('stream', False)
# 保留 Cursor 发送的原始模型名，响应时需要回填
cursor_model = payload.get('model', 'unknown')
msg_count = len(payload.get('messages', []))
ctx = build_route_context(client_model, is_stream)
# 容错：Responses 格式误入 CC 端点
if msg_count == 0 and 'input' in payload:
logger.info('检测到 Responses 格式（有 input 无 messages），自动转换')
payload = responses_to_cc(payload)
msg_count = len(payload.get('messages', []))
elif msg_count == 0:
logger.warning(f'messages 为空, payload keys: {list(payload.keys())}')
mapping = settings.resolve_model(cursor_model)
backend = mapping['backend']
upstream = mapping['upstream_model']
url_base = mapping['target_url']
api_key = mapping['api_key']
logger.info(
f'[CC] {cursor_model} → {upstream} '
f'后端={backend} 流式={is_stream} 消息数={msg_count}'
)
log_route_context('聊天补全', ctx, extra=f'消息数={message_count}')
_log_messages(payload)
if backend == 'openai':
return _via_openai(payload, upstream, url_base, api_key, is_stream, cursor_model)
else:
return _via_anthropic(payload, upstream, url_base, api_key, is_stream, cursor_model)
if ctx.backend == 'openai':
return _handle_openai_backend(ctx, payload)
if ctx.backend == 'responses':
return _handle_responses_backend(ctx, payload)
return _handle_anthropic_backend(ctx, payload)
# ─── OpenAI 后端 ──────────────────────────────────
def _normalize_chat_payload(payload: dict[str, Any]) -> tuple[dict[str, Any], int]:
"""整理聊天补全入口的请求体。
这里保留了一层兼容逻辑：当 Cursor 或调用方把 Responses 格式误发到
`/v1/chat/completions` 时，先降级转换成 Chat Completions，再进入统一主流程。
"""
message_count = len(payload.get('messages', []))
if message_count == 0 and 'input' in payload:
logger.info('检测到 Responses 格式误入聊天补全接口，已自动转换为 Chat Completions 格式')
payload = responses_to_cc(payload)
message_count = len(payload.get('messages', []))
elif message_count == 0:
logger.warning('消息列表为空，请求字段=%s', list(payload.keys()))
return payload, message_count
def _via_openai(payload, upstream, url_base, api_key, is_stream, cursor_model):
"""通过 OpenAI 兼容后端转发"""
_dbg(f'Cursor 原始请求 keys={list(payload.keys())} '
f'其他字段={json.dumps({k: v for k, v in payload.items() if k != "messages"}, ensure_ascii=False, default=str)[:500]}')
def _handle_openai_backend(ctx: RouteContext, payload: dict[str, Any]):
"""处理走 OpenAI 兼容后端的聊天补全请求。"""
_dbg(
'原始请求字段=' + str(list(payload.keys())) + ' '
+ '附加字段='
+ json.dumps(
{k: v for k, v in payload.items() if k != 'messages'},
ensure_ascii=False,
default=str,
)[:500]
)
payload = normalize_request(payload, upstream)
_dbg(f'normalize 后 model={payload.get("model")} tools数={len(payload.get("tools", []))}')
payload = normalize_request(payload, ctx.upstream_model)
_dbg(
f'标准化完成：模型={payload.get("model")} '
f'工具数={len(payload.get("tools", []))}'
)
headers = build_openai_headers(api_key)
url = f'{url_base.rstrip("/")}/v1/chat/completions'
url, headers = build_openai_target(ctx)
if not is_stream:
if ctx.is_stream:
return _handle_openai_stream(ctx, payload, url, headers)
return _handle_openai_non_stream(ctx, payload, url, headers)
def _handle_openai_non_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 OpenAI 兼容后端的非流式返回。"""
payload['stream'] = False
resp, err = forward_request(url, headers, payload)
if err:
return err
raw = resp.json()
_dbg(f'上游原始响应={json.dumps(raw, ensure_ascii=False, default=str)[:1000]}')
data = fix_response(raw)
data['model'] = cursor_model
_dbg(f'修复后响应={json.dumps(data, ensure_ascii=False, default=str)[:1000]}')
usage = data.get('usage', {})
logger.info(
f'[CC] 完成 prompt={usage.get("prompt_tokens", 0)} '
f'completion={usage.get("completion_tokens", 0)}'
)
return jsonify(data)
# 流式处理
raw = resp.json()
_dbg('上游原始响应=' + json.dumps(raw, ensure_ascii=False, default=str)[:1000])
data = fix_response(raw)
return _finalize_chat_response(ctx, data, debug_label='修复后响应')
def _handle_openai_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 OpenAI 兼容后端的流式返回。"""
payload['stream'] = True
_n = [0]
def generate():
resp, err = forward_request(url, headers, payload, stream=True)
if err:
yield f'data: {json.dumps({"error": {"message": err, "type": "upstream_error"}})}\n\n'
yield chat_error_chunk(str(err))
return
think_ext = ThinkTagExtractor()
think_extractor = ThinkTagExtractor()
chunk_count = 0
for chunk in iter_openai_sse(resp):
if chunk is None: # [DONE]
_dbg(f'流结束，共 {_n[0]} 个 chunk')
yield 'data: [DONE]\n\n'
if chunk is None:
_dbg(f'流式响应结束，共 {chunk_count} 个数据片段')
yield sse_data_message('[DONE]')
return
if _n[0] < 10:
_dbg(f'上游原始 chunk#{_n[0]}={json.dumps(chunk, ensure_ascii=False, default=str)[:500]}')
if chunk_count < 10:
_dbg(
f'上游原始片段#{chunk_count}='
+ json.dumps(chunk, ensure_ascii=False, default=str)[:500]
)
chunk = fix_stream_chunk(chunk)
chunk['model'] = cursor_model
chunk['model'] = ctx.client_model
for out in think_ext.process_chunk(chunk):
if _n[0] < 10:
_dbg(f'发给Cursor chunk#{_n[0]}={json.dumps(out, ensure_ascii=False, default=str)[:500]}')
yield f'data: {json.dumps(out)}\n\n'
for out in think_extractor.process_chunk(chunk):
if chunk_count < 10:
_dbg(
f'返回片段#{chunk_count}='
+ json.dumps(out, ensure_ascii=False, default=str)[:500]
)
yield sse_data_message(out)
_n[0] += 1
chunk_count += 1
return sse_response(generate())
# ─── Anthropic 后端 ───────────────────────────────
def _handle_responses_backend(ctx: RouteContext, payload: dict[str, Any]):
"""处理走原生 Responses 后端的聊天补全请求。
当上游只支持 `/v1/responses` 时，需要先把聊天补全请求转换为 Responses 请求，
返回时再转换回聊天补全协议。
"""
responses_payload = cc_to_responses_request(payload)
responses_payload['model'] = ctx.upstream_model
_dbg(
'已转换为 Responses 请求：字段=' + str(list(responses_payload.keys()))
+ f' 输入项数={len(responses_payload.get("input", []))}'
)
url, headers = build_responses_target(ctx)
if ctx.is_stream:
return _handle_responses_stream(ctx, responses_payload, url, headers)
return _handle_responses_non_stream(ctx, responses_payload, url, headers)
def _via_anthropic(payload, upstream, url_base, api_key, is_stream, cursor_model):
"""通过 Anthropic 后端转发（CC → Messages → CC）"""
payload['model'] = upstream
anthropic_payload = cc_to_messages_request(payload)
_dbg(f'CC→Messages 转换后 keys={list(anthropic_payload.keys())} '
f'messages数={len(anthropic_payload.get("messages", []))}')
headers = build_anthropic_headers(api_key)
url = f'{url_base.rstrip("/")}/v1/messages'
if not is_stream:
anthropic_payload['stream'] = False
resp, err = forward_request(url, headers, anthropic_payload)
def _handle_responses_non_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理原生 Responses 后端的非流式返回。"""
payload['stream'] = False
resp, err = forward_request(url, headers, payload)
if err:
return err
raw = resp.json()
_dbg(f'上游原始响应={json.dumps(raw, ensure_ascii=False, default=str)[:1000]}')
data = messages_to_cc_response(raw)
data['model'] = cursor_model
_dbg(f'Messages→CC 转换后={json.dumps(data, ensure_ascii=False, default=str)[:1000]}')
usage = data.get('usage', {})
logger.info(
f'[CC] 完成 prompt={usage.get("prompt_tokens", 0)} '
f'completion={usage.get("completion_tokens", 0)}'
)
return jsonify(data)
# 流式处理
anthropic_payload['stream'] = True
converter = AnthropicStreamConverter()
_n = [0]
raw = resp.json()
_dbg('上游原始响应=' + json.dumps(raw, ensure_ascii=False, default=str)[:1000])
data = responses_to_cc_response(raw, ctx.client_model)
return _finalize_chat_response(ctx, data, debug_label='Responses 转回聊天补全后')
def _handle_responses_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理原生 Responses 后端的流式返回。"""
payload['stream'] = True
converter = ResponsesToCCStreamConverter(model=ctx.client_model)
def generate():
resp, err = forward_request(url, headers, anthropic_payload, stream=True)
resp, err = forward_request(url, headers, payload, stream=True)
if err:
yield f'data: {json.dumps({"error": {"message": err, "type": "upstream_error"}})}\n\n'
yield chat_error_chunk(str(err))
return
event_count = 0
for event_type, event_data in iter_responses_sse(resp):
if event_count < 10:
_dbg(
f'上游事件#{event_count} 类型={event_type} 数据='
+ json.dumps(event_data, ensure_ascii=False, default=str)[:500]
)
for chunk in converter.process_event(event_type, event_data):
if event_count < 10:
_dbg(
f'返回片段#{event_count}='
+ json.dumps(chunk, ensure_ascii=False, default=str)[:500]
)
yield sse_data_message(chunk)
event_count += 1
_dbg(f'流式响应结束，共 {event_count} 个事件')
yield sse_data_message('[DONE]')
return sse_response(generate())
def _handle_anthropic_backend(ctx: RouteContext, payload: dict[str, Any]):
"""处理走 Anthropic Messages 后端的聊天补全请求。"""
payload['model'] = ctx.upstream_model
anthropic_payload = cc_to_messages_request(payload)
_dbg(
'已转换为 Messages 请求：字段=' + str(list(anthropic_payload.keys()))
+ f' 消息数={len(anthropic_payload.get("messages", []))}'
)
url, headers = build_anthropic_target(ctx)
if ctx.is_stream:
return _handle_anthropic_stream(ctx, anthropic_payload, url, headers)
return _handle_anthropic_non_stream(ctx, anthropic_payload, url, headers)
def _handle_anthropic_non_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 Anthropic 后端的非流式返回。"""
payload['stream'] = False
resp, err = forward_request(url, headers, payload)
if err:
return err
raw = resp.json()
_dbg('上游原始响应=' + json.dumps(raw, ensure_ascii=False, default=str)[:1000])
data = messages_to_cc_response(raw)
return _finalize_chat_response(ctx, data, debug_label='Messages 转回聊天补全后')
def _handle_anthropic_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 Anthropic 后端的流式返回。
这里仍然保留独立的事件级转换器，而不是先落成完整响应再回放，
是为了尽量保持 Cursor 端的流式体验和工具调用时序。
"""
payload['stream'] = True
converter = AnthropicStreamConverter()
def generate():
resp, err = forward_request(url, headers, payload, stream=True)
if err:
yield chat_error_chunk(str(err))
return
event_count = 0
for event_type, event_data in iter_anthropic_sse(resp):
if _n[0] < 10:
_dbg(f'上游事件#{_n[0]} {event_type}={json.dumps(event_data, ensure_ascii=False, default=str)[:500]}')
if event_count < 10:
_dbg(
f'上游事件#{event_count} 类型={event_type} 数据='
+ json.dumps(event_data, ensure_ascii=False, default=str)[:500]
)
for chunk_str in converter.process_event(event_type, event_data):
try:
chunk_obj = json.loads(chunk_str)
chunk_obj['model'] = cursor_model
chunk_str = json.dumps(chunk_obj)
chunk_obj['model'] = ctx.client_model
chunk_str = json.dumps(chunk_obj, ensure_ascii=False)
except (json.JSONDecodeError, TypeError):
pass
if _n[0] < 10:
_dbg(f'发给Cursor chunk#{_n[0]}={chunk_str[:500]}')
yield f'data: {chunk_str}\n\n'
_n[0] += 1
if event_count < 10:
_dbg(f'返回片段#{event_count}={chunk_str[:500]}')
yield sse_data_message(chunk_str)
_dbg(f'流结束，共 {_n[0]} 个事件')
yield 'data: [DONE]\n\n'
event_count += 1
_dbg(f'流式响应结束，共 {event_count} 个事件')
yield sse_data_message('[DONE]')
return sse_response(generate())
def _log_messages(payload):
"""记录请求中的消息摘要"""
for i, msg in enumerate(payload.get('messages', [])):
role = msg.get('role', '?')
content = msg.get('content')
def _finalize_chat_response(
ctx: RouteContext,
data: dict[str, Any],
*,
debug_label: str,
):
"""统一收尾非流式聊天补全响应。
三条后端链路最终都会回到 Chat Completions 格式，因此这里集中做：
- 回填给 Cursor 展示的模型名
- 输出统一调试日志
- 输出统一令牌统计日志
"""
data['model'] = ctx.client_model
_dbg(debug_label + '=' + json.dumps(data, ensure_ascii=False, default=str)[:1000])
log_usage('聊天补全', data.get('usage', {}), input_key='prompt_tokens', output_key='completion_tokens')
return jsonify(data)
def _log_messages(payload: dict[str, Any]) -> None:
"""记录消息摘要，方便排查请求形态是否符合预期。"""
for index, message in enumerate(payload.get('messages', [])):
role = message.get('role', '?')
content = message.get('content')
extra = ''
if 'tool_calls' in msg:
extra += f' tool_calls={len(msg["tool_calls"])}'
if msg.get('tool_call_id'):
extra += f' tool_call_id={msg["tool_call_id"]}'
if 'tool_calls' in message:
extra += f' 工具调用数={len(message["tool_calls"])}'
if message.get('tool_call_id'):
extra += f' 工具调用ID={message["tool_call_id"]}'
if isinstance(content, list):
info = f'list[{len(content)}]'
content_info = f'列表[{len(content)}]'
elif isinstance(content, str):
info = f'str[{len(content)}]'
content_info = f'文本[{len(content)}]'
else:
info = type(content).__name__
logger.info(f' 消息[{i}] {role} {info}{extra}')
content_info = type(content).__name__
logger.info(' 消息[%s] 角色=%s 内容=%s%s', index, role, content_info, extra)

118
routes/common.py Normal file
View file

@ -0,0 +1,118 @@
"""路由层公共辅助
收敛多个数据面路由都会用到的上下文解析、上游目标构造、日志输出和
SSE 消息拼装逻辑，避免 `chat.py` 和 `responses.py` 各自维护重复实现。
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import logging
from typing import Any
import settings
from utils.http import build_anthropic_headers, build_openai_headers
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class RouteContext:
"""数据面路由使用的标准请求上下文。"""
client_model: str
upstream_model: str
backend: str
target_url: str
api_key: str
is_stream: bool
def build_route_context(client_model: str, is_stream: bool) -> RouteContext:
"""解析模型映射，得到当前请求的统一路由上下文。"""
mapping = settings.resolve_model(client_model)
return RouteContext(
client_model=client_model,
upstream_model=mapping['upstream_model'],
backend=mapping['backend'],
target_url=mapping['target_url'],
api_key=mapping['api_key'],
is_stream=is_stream,
)
def build_openai_target(ctx: RouteContext) -> tuple[str, dict[str, str]]:
"""根据路由上下文生成 OpenAI 兼容后端的地址和请求头。"""
url = f'{ctx.target_url.rstrip("/")}/v1/chat/completions'
headers = build_openai_headers(ctx.api_key)
return url, headers
def build_responses_target(ctx: RouteContext) -> tuple[str, dict[str, str]]:
"""根据路由上下文生成 OpenAI Responses 后端的地址和请求头。"""
url = f'{ctx.target_url.rstrip("/")}/v1/responses'
headers = build_openai_headers(ctx.api_key)
return url, headers
def build_anthropic_target(ctx: RouteContext) -> tuple[str, dict[str, str]]:
"""根据路由上下文生成 Anthropic 后端的地址和请求头。"""
url = f'{ctx.target_url.rstrip("/")}/v1/messages'
headers = build_anthropic_headers(ctx.api_key)
return url, headers
def log_route_context(route_name: str, ctx: RouteContext, *, extra: str = '') -> None:
"""统一输出路由级日志，避免不同入口的日志格式逐渐漂移。"""
parts = [
f'[{route_name}]',
f'模型={ctx.client_model}',
f'上游模型={ctx.upstream_model}',
f'后端={ctx.backend}',
f'流式={ctx.is_stream}',
]
if extra:
parts.append(extra)
logger.info(' '.join(parts))
def log_usage(
route_name: str,
usage: dict[str, Any],
*,
input_key: str,
output_key: str,
) -> None:
"""统一输出令牌统计日志。
不同协议对 usage 字段命名不一致，这里只接收字段名，不在调用方重复拼接日志文案。
"""
logger.info(
'[%s] 请求完成 输入令牌=%s 输出令牌=%s',
route_name,
usage.get(input_key, 0),
usage.get(output_key, 0),
)
def sse_data_message(data: Any) -> str:
"""构造仅包含 data 的 SSE 消息。"""
payload = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False)
return f'data: {payload}\n\n'
def sse_event_message(event_type: str, data: Any) -> str:
"""构造带 event 名称的 SSE 消息。"""
payload = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False)
return f'event: {event_type}\ndata: {payload}\n\n'
def chat_error_chunk(message: str, error_type: str = 'upstream_error') -> str:
"""构造聊天补全流式接口使用的错误消息。"""
return sse_data_message({'error': {'message': message, 'type': error_type}})
def responses_error_event(message: str) -> str:
"""构造 Responses 流式接口使用的错误事件。"""
return sse_event_message('error', {'error': message})

285
routes/responses.py
View file

@ -1,22 +1,37 @@
"""路由: /v1/responses
处理 Cursor 对 GPT/Claude-Opus 等模型发出的 Responses API 格式请求。
转换为 CC 格式后分发到对应后端，响应再转回 Responses 格式。
处理 Cursor 对 GPT、Claude-Opus 等模型发出的 Responses API 请求。
请求会先转换为 Chat Completions 中间表示，再按后端类型分发，最后转换回 Responses 格式。
"""
from __future__ import annotations
import json
import logging
from typing import Any
from flask import Blueprint, request, jsonify
from flask import Blueprint, jsonify, request
import settings
from adapters.responses_adapter import responses_to_cc, cc_to_responses, ResponsesStreamConverter
from adapters.openai_fixer import normalize_request, fix_response, fix_stream_chunk
from adapters.openai_anthropic import cc_to_messages_request, messages_to_cc_response
from adapters.cc_anthropic_adapter import cc_to_messages_request, messages_to_cc_response
from adapters.openai_compat_fixer import fix_response, fix_stream_chunk, normalize_request
from adapters.responses_cc_adapter import ResponsesStreamConverter, cc_to_responses, responses_to_cc
from config import Config
from routes.common import (
RouteContext,
build_anthropic_target,
build_openai_target,
build_responses_target,
build_route_context,
log_route_context,
log_usage,
responses_error_event,
)
from utils.http import (
build_openai_headers, build_anthropic_headers,
forward_request, sse_response,
iter_openai_sse, iter_anthropic_sse,
forward_request,
iter_anthropic_sse,
iter_openai_sse,
iter_responses_sse,
sse_response,
)
from utils.think_tag import ThinkTagExtractor
@ -25,102 +40,272 @@ logger = logging.getLogger(__name__)
bp = Blueprint('responses', __name__)
def _dbg(message: str) -> None:
"""仅在调试模式下输出详细日志。"""
if Config.DEBUG:
logger.info('[响应生成调试] %s', message)
@bp.route('/v1/responses', methods=['POST'])
def responses_endpoint():
"""处理 Responses 请求并按模型映射分发。"""
payload = request.get_json(force=True)
model = payload.get('model', 'unknown')
client_model = payload.get('model', 'unknown')
is_stream = payload.get('stream', False)
mapping = settings.resolve_model(model)
backend = mapping['backend']
upstream = mapping['upstream_model']
url_base = mapping['target_url']
api_key = mapping['api_key']
ctx = build_route_context(client_model, is_stream)
log_route_context('响应生成', ctx)
logger.info(f'[Responses] {model} → {upstream} 后端={backend} 流式={is_stream}')
cc_payload = _build_cc_payload(payload, ctx)
# Responses → CC
if ctx.backend == 'openai':
return _handle_openai_backend(ctx, cc_payload)
if ctx.backend == 'responses':
return _handle_responses_backend(ctx, payload)
return _handle_anthropic_backend(ctx, cc_payload)
def _build_cc_payload(payload: dict[str, Any], ctx: RouteContext) -> dict[str, Any]:
"""将 Responses 请求统一降级为 Chat Completions 中间表示。
这样后续无论走 OpenAI 兼容后端还是 Anthropic 后端，都能复用一套
中间协议，避免在路由层同时维护两套完全不同的请求编排逻辑。
"""
cc_payload = responses_to_cc(payload)
cc_payload['model'] = upstream
if backend == 'openai':
return _via_openai(cc_payload, url_base, api_key, is_stream, model)
else:
return _via_anthropic(cc_payload, url_base, api_key, is_stream, model)
cc_payload['model'] = ctx.upstream_model
_dbg(
'已转换为聊天补全中间表示：字段=' + str(list(cc_payload.keys()))
+ f' 消息数={len(cc_payload.get("messages", []))}'
)
return cc_payload
# ─── OpenAI 后端 ──────────────────────────────────
def _via_openai(cc_payload, url_base, api_key, is_stream, display_model):
"""通过 OpenAI 后端处理"""
def _handle_openai_backend(ctx: RouteContext, cc_payload: dict[str, Any]):
"""处理走 OpenAI 兼容后端的 Responses 请求。"""
cc_payload = normalize_request(cc_payload)
headers = build_openai_headers(api_key)
url = f'{url_base.rstrip("/")}/v1/chat/completions'
_dbg(
f'标准化完成：模型={cc_payload.get("model")} '
f'工具数={len(cc_payload.get("tools", []))}'
)
if not is_stream:
url, headers = build_openai_target(ctx)
if ctx.is_stream:
return _handle_openai_stream(ctx, cc_payload, url, headers)
return _handle_openai_non_stream(ctx, cc_payload, url, headers)
def _handle_openai_non_stream(
ctx: RouteContext,
cc_payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 OpenAI 兼容后端的非流式 Responses 返回。"""
cc_payload['stream'] = False
resp, err = forward_request(url, headers, cc_payload)
if err:
return err
return jsonify(cc_to_responses(fix_response(resp.json()), display_model))
# 流式处理
raw = resp.json()
_dbg('上游原始响应=' + json.dumps(raw, ensure_ascii=False, default=str)[:1000])
fixed = fix_response(raw)
response_data = cc_to_responses(fixed, ctx.client_model)
return _finalize_responses_response(response_data, debug_label='转换为 Responses 后')
def _handle_openai_stream(
ctx: RouteContext,
cc_payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 OpenAI 兼容后端的流式 Responses 返回。"""
cc_payload['stream'] = True
converter = ResponsesStreamConverter(model=display_model)
converter = ResponsesStreamConverter(model=ctx.client_model)
def generate():
yield from converter.start_events()
resp, err = forward_request(url, headers, cc_payload, stream=True)
if err:
yield f'event: error\ndata: {json.dumps({"error": err})}\n\n'
yield responses_error_event(str(err))
return
think_ext = ThinkTagExtractor()
think_extractor = ThinkTagExtractor()
chunk_count = 0
for chunk in iter_openai_sse(resp):
if chunk is None:
_dbg(f'流式响应结束，共 {chunk_count} 个数据片段')
yield from converter.finalize()
return
if chunk_count < 10:
_dbg(
f'上游原始片段#{chunk_count}='
+ json.dumps(chunk, ensure_ascii=False, default=str)[:500]
)
chunk = fix_stream_chunk(chunk)
for out in think_ext.process_chunk(chunk):
for out in think_extractor.process_chunk(chunk):
if chunk_count < 10:
_dbg(
f'转换后片段#{chunk_count}='
+ json.dumps(out, ensure_ascii=False, default=str)[:500]
)
yield from converter.process_cc_chunk(out)
chunk_count += 1
return sse_response(generate())
# ─── Anthropic 后端 ───────────────────────────────
def _handle_responses_backend(ctx: RouteContext, payload: dict[str, Any]):
"""处理走原生 Responses 后端的请求。
当中转站本身就只支持 `/v1/responses` 时，不需要再绕到聊天补全中间协议，
直接转发原生 Responses 请求即可。
"""
payload = dict(payload)
payload['model'] = ctx.upstream_model
url, headers = build_responses_target(ctx)
if ctx.is_stream:
return _handle_responses_stream(ctx, payload, url, headers)
return _handle_responses_non_stream(ctx, payload, url, headers)
def _via_anthropic(cc_payload, url_base, api_key, is_stream, display_model):
"""通过 Anthropic 后端处理"""
def _handle_responses_non_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理原生 Responses 后端的非流式返回。"""
payload['stream'] = False
resp, err = forward_request(url, headers, payload)
if err:
return err
response_data = resp.json()
response_data['model'] = ctx.client_model
return _finalize_responses_response(response_data, debug_label='原生 Responses 返回后')
def _handle_responses_stream(
ctx: RouteContext,
payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理原生 Responses 后端的流式返回。"""
payload['stream'] = True
converter = ResponsesStreamConverter(model=ctx.client_model)
def generate():
resp, err = forward_request(url, headers, payload, stream=True)
if err:
yield responses_error_event(str(err))
return
event_count = 0
for event_type, event_data in iter_responses_sse(resp):
if event_count < 10:
_dbg(
f'上游事件#{event_count} 类型={event_type} 数据='
+ json.dumps(event_data, ensure_ascii=False, default=str)[:500]
)
yield from converter.process_responses_event(event_type, event_data)
event_count += 1
_dbg(f'流式响应结束，共 {event_count} 个事件')
return sse_response(generate())
def _handle_anthropic_backend(ctx: RouteContext, cc_payload: dict[str, Any]):
"""处理走 Anthropic 后端的 Responses 请求。"""
anthropic_payload = cc_to_messages_request(cc_payload)
headers = build_anthropic_headers(api_key)
url = f'{url_base.rstrip("/")}/v1/messages'
_dbg(
'已转换为 Messages 请求：字段=' + str(list(anthropic_payload.keys()))
+ f' 消息数={len(anthropic_payload.get("messages", []))}'
)
if not is_stream:
url, headers = build_anthropic_target(ctx)
if ctx.is_stream:
return _handle_anthropic_stream(ctx, anthropic_payload, url, headers)
return _handle_anthropic_non_stream(ctx, anthropic_payload, url, headers)
def _handle_anthropic_non_stream(
ctx: RouteContext,
anthropic_payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 Anthropic 后端的非流式 Responses 返回。"""
anthropic_payload['stream'] = False
resp, err = forward_request(url, headers, anthropic_payload)
if err:
return err
cc_data = messages_to_cc_response(resp.json())
return jsonify(cc_to_responses(cc_data, display_model))
# 流式处理：Anthropic SSE → Responses SSE（跳过 CC 中间态）
raw = resp.json()
_dbg('上游原始响应=' + json.dumps(raw, ensure_ascii=False, default=str)[:1000])
cc_data = messages_to_cc_response(raw)
response_data = cc_to_responses(cc_data, ctx.client_model)
return _finalize_responses_response(response_data, debug_label='Messages 转回 Responses 后')
def _handle_anthropic_stream(
ctx: RouteContext,
anthropic_payload: dict[str, Any],
url: str,
headers: dict[str, str],
):
"""处理 Anthropic 后端的流式 Responses 返回。
这里直接将 Anthropic SSE 事件映射到 Responses SSE，故意跳过 CC 流式中间态，
这样可以减少一次事件重组，降低流式转换复杂度，也更容易保留原始时序。
"""
anthropic_payload['stream'] = True
converter = ResponsesStreamConverter(model=display_model)
converter = ResponsesStreamConverter(model=ctx.client_model)
def generate():
yield from converter.start_events()
resp, err = forward_request(url, headers, anthropic_payload, stream=True)
if err:
yield f'event: error\ndata: {json.dumps({"error": err})}\n\n'
yield responses_error_event(str(err))
return
event_count = 0
for event_type, event_data in iter_anthropic_sse(resp):
yield from converter.process_anthropic_event(event_type, event_data)
if event_count < 10:
_dbg(
f'上游事件#{event_count} 类型={event_type} 数据='
+ json.dumps(event_data, ensure_ascii=False, default=str)[:500]
)
yield from converter.process_anthropic_event(event_type, event_data)
event_count += 1
_dbg(f'流式响应结束，共 {event_count} 个事件')
yield from converter.finalize()
return sse_response(generate())
def _finalize_responses_response(response_data: dict[str, Any], *, debug_label: str):
"""统一收尾非流式 Responses 响应。
两条转换链路和一条原生 Responses 链路最终都会回到 Responses 对象，因此这里集中
处理调试日志、回填展示模型名以及 usage 日志。
"""
response_data['model'] = response_data.get('model') or ''
_dbg(debug_label + '=' + json.dumps(response_data, ensure_ascii=False, default=str)[:1000])
log_usage('响应生成', response_data.get('usage', {}), input_key='input_tokens', output_key='output_tokens')
return jsonify(response_data)

5
settings.py
View file

@ -76,9 +76,12 @@ def resolve_model(model_name):
if model_name in mappings:
m = mappings[model_name]
backend = m.get('backend')
if backend in ('', None, 'auto'):
backend = _auto_detect(model_name)
return {
'upstream_model': m.get('upstream_model') or model_name,
'backend': m.get('backend') or _auto_detect(model_name),
'backend': backend,
'target_url': m.get('target_url') or base_url,
'api_key': m.get('api_key') or base_key,
}

1
static/admin.css
View file

@ -58,6 +58,7 @@ main{padding:28px 0 60px}
.tag{font-size:11px;padding:2px 8px;border-radius:4px;font-weight:500}
.tag-anthropic{background:rgba(249,115,22,.15);color:#fb923c}
.tag-openai{background:rgba(16,185,129,.15);color:#34d399}
.tag-responses{background:rgba(59,130,246,.15);color:#60a5fa}
.tag-auto{background:rgba(139,92,246,.15);color:#a78bfa}
.tag-override{background:rgba(59,130,246,.1);color:var(--primary)}
.mapping-actions{margin-left:auto;display:flex;gap:6px}

2
static/admin.html
View file

@ -94,11 +94,13 @@
<option value="auto">自动检测</option>
<option value="anthropic">Anthropic (/v1/messages)</option>
<option value="openai">OpenAI (/v1/chat/completions)</option>
<option value="responses">OpenAI Responses (/v1/responses)</option>
</select>
</div>
<div class="hint">
<b>anthropic</b>：转换为 Anthropic Messages 格式 — 适用于中转站通过 <code>/v1/messages</code> 提供 Claude 模型<br>
<b>openai</b>：保持 OpenAI Chat Completions 格式 — 适用于 GPT、DeepSeek、Codex 或通过 <code>/v1/chat/completions</code> 提供所有模型的中转站<br>
<b>responses</b>：保持 OpenAI Responses 格式 — 适用于中转站仅通过 <code>/v1/responses</code> 提供模型能力<br>
<b>自动检测</b>：根据上游模型名判断（含 claude → anthropic，其他 → openai）
</div>
</div>

14
static/admin.js
View file

@ -124,8 +124,18 @@ async function loadMappings() {
el.innerHTML = '<div class="mapping-list">' + keys.map(name => {
const m = mappings[name];
const backend = m.backend || 'auto';
const tagClass = backend === 'anthropic' ? 'tag-anthropic' : backend === 'openai' ? 'tag-openai' : 'tag-auto';
const tagLabel = backend === 'auto' ? '自动' : backend;
const tagClass = backend === 'anthropic'
? 'tag-anthropic'
: backend === 'responses'
? 'tag-responses'
: backend === 'openai'
? 'tag-openai'
: 'tag-auto';
const tagLabel = backend === 'auto'
? '自动'
: backend === 'responses'
? 'responses'
: backend;
const hasOverride = m.target_url || m.api_key;
return `<div class="mapping-item">
<div class="mapping-top">

25
utils/http.py
View file

@ -1,8 +1,11 @@
"""HTTP 工具 - 请求头构建、上游转发、SSE 流解析、响应构建"""
from __future__ import annotations
import json
import uuid
import logging
import uuid
from typing import Any, Iterator
import requests
from flask import Response, jsonify
@ -12,7 +15,7 @@ from config import Config
logger = logging.getLogger(__name__)
def gen_id(prefix=''):
def gen_id(prefix: str = '') -> str:
"""生成唯一 ID"""
return f'{prefix}{uuid.uuid4().hex[:24]}'
@ -20,7 +23,7 @@ def gen_id(prefix=''):
# ─── 请求头构建 ────────────────────────────────────
def build_openai_headers(api_key):
def build_openai_headers(api_key: str) -> dict[str, str]:
"""构建 OpenAI 兼容请求头"""
return {
'Authorization': f'Bearer {api_key}',
@ -28,7 +31,7 @@ def build_openai_headers(api_key):
}
def build_anthropic_headers(api_key):
def build_anthropic_headers(api_key: str) -> dict[str, str]:
"""构建 Anthropic 请求头，根据密钥前缀自动选择鉴权方式"""
headers = {
'anthropic-version': '2023-06-01',
@ -94,7 +97,7 @@ def forward_request(url, headers, payload, stream=False):
# ─── SSE 流解析 ───────────────────────────────────
def iter_openai_sse(response):
def iter_openai_sse(response) -> Iterator[dict[str, Any] | None]:
"""解析 OpenAI SSE 流，yield chunk 字典；yield None 表示 [DONE]"""
for line in response.iter_lines():
if not line:
@ -112,8 +115,18 @@ def iter_openai_sse(response):
continue
def iter_anthropic_sse(response):
def iter_anthropic_sse(response) -> Iterator[tuple[str, dict[str, Any]]]:
"""解析 Anthropic SSE 流，yield (event_type, data_dict) 元组"""
yield from _iter_event_sse(response)
def iter_responses_sse(response) -> Iterator[tuple[str, dict[str, Any]]]:
"""解析 OpenAI Responses SSE 流，yield (event_type, data_dict) 元组"""
yield from _iter_event_sse(response)
def _iter_event_sse(response) -> Iterator[tuple[str, dict[str, Any]]]:
"""解析带 event/data 的通用 SSE 流。"""
event_type = ''
for line in response.iter_lines():
if not line: