| from __future__ import annotations |
|
|
| import json |
| import os |
| import urllib.error |
| import urllib.request |
| import uuid |
| from dataclasses import dataclass |
|
|
|
|
| @dataclass |
| class CodexResult: |
| final_response: str |
| finish_reason: str = "completed" |
|
|
|
|
| class CodexHarness: |
| """Small non-interactive adapter for OpenCode Go's Responses API.""" |
|
|
| def __init__(self, model=None, api_key=None, base_url=None, |
| session_root=None, system_prompt=None, **kwargs): |
| self.model = model or os.environ.get("OPENCODE_GO_MODEL", "gpt-5.6-luna") |
| self.api_key = ( |
| api_key |
| or os.environ.get("OPENCODE_GO_API_KEY") |
| or os.environ.get("OPENAI_API_KEY") |
| or "" |
| ).strip() |
| self.base_url = ( |
| base_url |
| or os.environ.get("OPENCODE_GO_BASE_URL") |
| or os.environ.get("OPENAI_BASE_URL") |
| or "https://opencode.ai/zen/go/v1" |
| ).rstrip("/") |
| self.system_prompt = system_prompt or "" |
|
|
| @staticmethod |
| def _text(payload: dict) -> str: |
| text = str(payload.get("output_text") or "").strip() |
| if text: |
| return text |
| parts = [] |
| for item in payload.get("output") or []: |
| for content in item.get("content") or []: |
| if content.get("type") in {"output_text", "text"}: |
| value = content.get("text") |
| if isinstance(value, str): |
| parts.append(value) |
| elif isinstance(value, dict) and isinstance(value.get("value"), str): |
| parts.append(value["value"]) |
| return "".join(parts).strip() |
|
|
| def run(self, prompt, session_id=None): |
| if not self.api_key: |
| raise RuntimeError("OPENCODE_GO_API_KEY is not configured") |
| full_prompt = (self.system_prompt + "\n\n" + prompt).strip() |
| body = json.dumps({"model": self.model, "input": full_prompt}, ensure_ascii=False).encode("utf-8") |
| request = urllib.request.Request( |
| self.base_url + "/responses", |
| data=body, |
| method="POST", |
| headers={ |
| "Authorization": "Bearer " + self.api_key, |
| "Content-Type": "application/json", |
| "Accept": "application/json", |
| "User-Agent": "Global-Marine-Foundation/4.4", |
| "x-opencode-session": str(session_id or uuid.uuid4()), |
| }, |
| ) |
| try: |
| with urllib.request.urlopen(request, timeout=600) as response: |
| payload = json.loads(response.read().decode("utf-8")) |
| except urllib.error.HTTPError as exc: |
| detail = exc.read().decode("utf-8", errors="replace")[:1000] |
| raise RuntimeError(f"OpenCode Go API {exc.code}: {detail}") from exc |
| except urllib.error.URLError as exc: |
| raise RuntimeError(f"OpenCode Go connection failed: {exc.reason}") from exc |
| answer = self._text(payload) |
| if not answer: |
| raise RuntimeError("OpenCode Go returned no text: " + json.dumps(payload, ensure_ascii=False)[:800]) |
| return CodexResult(final_response=answer) |
|
|
