OpenAI Responses API for Codex CLI compatibility.
(self, body: bytes)
| 595 | }) |
| 596 | |
| 597 | def handle_responses(self, body: bytes): |
| 598 | """OpenAI Responses API for Codex CLI compatibility.""" |
| 599 | req = json.loads(body) |
| 600 | model_name, model_id, think_mode, err = self._resolve_model( |
| 601 | req.get("model", CONFIG["default_model"])) |
| 602 | if err: |
| 603 | self.send_json({"error": {"message": err}}, 400) |
| 604 | return |
| 605 | |
| 606 | input_items = req.get("input", []) |
| 607 | tools = req.get("tools") |
| 608 | |
| 609 | messages = [] |
| 610 | if req.get("instructions"): |
| 611 | messages.append({"role": "system", "content": req["instructions"]}) |
| 612 | if isinstance(input_items, str): |
| 613 | messages.append({"role": "user", "content": input_items}) |
| 614 | elif isinstance(input_items, list): |
| 615 | for item in input_items: |
| 616 | if isinstance(item, str): |
| 617 | messages.append({"role": "user", "content": item}) |
| 618 | elif isinstance(item, dict): |
| 619 | if item.get("type") == "function_call_output": |
| 620 | messages.append({"role": "tool", "tool_call_id": item.get("call_id", ""), |
| 621 | "name": item.get("name", ""), "content": item.get("output", "")}) |
| 622 | elif item.get("role") == "assistant" or (item.get("type") == "message" and item.get("role") == "assistant"): |
| 623 | cp = item.get("content", []) |
| 624 | text_acc, tc_list = "", [] |
| 625 | if isinstance(cp, list): |
| 626 | for c in cp: |
| 627 | if isinstance(c, dict): |
| 628 | if c.get("type") == "output_text": text_acc += c.get("text", "") |
| 629 | elif c.get("type") == "function_call": tc_list.append(c) |
| 630 | elif isinstance(cp, str): |
| 631 | text_acc = cp |
| 632 | m = {"role": "assistant", "content": text_acc or None} |
| 633 | if tc_list: |
| 634 | m["tool_calls"] = [{"id": tc.get("call_id", f"call_{i}"), "type": "function", |
| 635 | "function": {"name": tc.get("name",""), "arguments": tc.get("arguments","{}")}} |
| 636 | for i, tc in enumerate(tc_list)] |
| 637 | messages.append(m) |
| 638 | else: |
| 639 | role = item.get("role", "user") |
| 640 | content = item.get("content", "") |
| 641 | if isinstance(content, list): |
| 642 | content = " ".join(c.get("text", "") for c in content if c.get("type") in ("text", "input_text")) |
| 643 | messages.append({"role": role, "content": content}) |
| 644 | |
| 645 | if tools: |
| 646 | tools = [{"type": "function", "function": {"name": t["name"], "description": t.get("description", ""), "parameters": t.get("parameters", {})}} |
| 647 | if t.get("type") == "function" and "function" not in t else t for t in tools] |
| 648 | |
| 649 | prompt = messages_to_prompt(messages, tools) |
| 650 | if not prompt.strip(): |
| 651 | self.send_json({"error": {"message": "empty input"}}, 400) |
| 652 | return |
| 653 | |
| 654 | try: |
no test coverage detected