(self, body: bytes)
| 204 | # ─── /v1/responses (Codex CLI) ─────────────────────────────────────────── |
| 205 | |
| 206 | def _handle_responses(self, body: bytes): |
| 207 | req = self._parse_body(body) |
| 208 | if req is None: |
| 209 | self.send_json({"error": {"message": "invalid JSON"}}, 400) |
| 210 | return |
| 211 | model_name, model_id, think_mode, err, extra_fields = resolve_model( |
| 212 | req.get("model", CONFIG["default_model"])) |
| 213 | if err: |
| 214 | self.send_json({"error": {"message": err}}, 400) |
| 215 | return |
| 216 | |
| 217 | input_items = req.get("input", []) |
| 218 | tools = req.get("tools") |
| 219 | messages = [] |
| 220 | if req.get("instructions"): |
| 221 | messages.append({"role": "system", "content": req["instructions"]}) |
| 222 | if isinstance(input_items, str): |
| 223 | messages.append({"role": "user", "content": input_items}) |
| 224 | elif isinstance(input_items, list): |
| 225 | for item in input_items: |
| 226 | if isinstance(item, str): |
| 227 | messages.append({"role": "user", "content": item}) |
| 228 | elif isinstance(item, dict): |
| 229 | if item.get("type") == "function_call_output": |
| 230 | messages.append({"role": "tool", "tool_call_id": item.get("call_id", ""), |
| 231 | "name": item.get("name", ""), "content": item.get("output", "")}) |
| 232 | elif item.get("role") == "assistant" or (item.get("type") == "message" and item.get("role") == "assistant"): |
| 233 | cp = item.get("content", []) |
| 234 | text_acc, tc_list = "", [] |
| 235 | if isinstance(cp, list): |
| 236 | for c in cp: |
| 237 | if isinstance(c, dict): |
| 238 | if c.get("type") == "output_text": text_acc += c.get("text", "") |
| 239 | elif c.get("type") == "function_call": tc_list.append(c) |
| 240 | elif isinstance(cp, str): |
| 241 | text_acc = cp |
| 242 | m = {"role": "assistant", "content": text_acc or None} |
| 243 | if tc_list: |
| 244 | m["tool_calls"] = [{"id": tc.get("call_id", f"call_{i}"), "type": "function", |
| 245 | "function": {"name": tc.get("name",""), "arguments": tc.get("arguments","{}")}} |
| 246 | for i, tc in enumerate(tc_list)] |
| 247 | messages.append(m) |
| 248 | else: |
| 249 | role = item.get("role", "user") |
| 250 | content = item.get("content", "") |
| 251 | if isinstance(content, list): |
| 252 | content = " ".join(c.get("text", "") for c in content if c.get("type") in ("text", "input_text")) |
| 253 | messages.append({"role": role, "content": content}) |
| 254 | |
| 255 | if tools: |
| 256 | tools = [{"type": "function", "function": {"name": t["name"], "description": t.get("description", ""), "parameters": t.get("parameters", {})}} |
| 257 | if t.get("type") == "function" and "function" not in t else t for t in tools] |
| 258 | |
| 259 | tool_choice = req.get("tool_choice", "auto") |
| 260 | prompt, images = messages_to_prompt(messages, tools, tool_choice) |
| 261 | if not prompt.strip(): |
| 262 | self.send_json({"error": {"message": "empty input"}}, 400) |
| 263 | return |
no test coverage detected