# devinx — Claude Code on Cognition SWE-2 via cliproxyapi (setup guide)

> Source: <https://gist.github.com/future3OOO/9d829db4dda0a5f539028236c7091767>
> Published: 2026-09-11 05:36:01+00:00

|  | #!/usr/bin/env python3 | 
|  | """OpenAI-compatible shim in front of Cognition's Connect-RPC GetChatMessage. | 
|  |  | 
|  | Serves POST /v1/chat/completions (stream + non-stream) and GET /v1/models by | 
|  | translating to exa.api_server_pb.ApiServerService/GetChatMessage against | 
|  | server.codeium.com, authenticated with the Devin CLI's stored credential. | 
|  | Runs behind CLIProxyAPI as an openai-compatibility upstream. | 
|  | """ | 
|  | import glob | 
|  | import gzip | 
|  | import json | 
|  | import os | 
|  | import re | 
|  | import struct | 
|  | import threading | 
|  | import time | 
|  | import uuid | 
|  | from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer | 
|  |  | 
|  | import requests | 
|  | from google.protobuf import descriptor_pb2, descriptor_pool, message_factory | 
|  | from google.protobuf import timestamp_pb2, duration_pb2, any_pb2, struct_pb2 | 
|  | from google.protobuf import wrappers_pb2, empty_pb2, field_mask_pb2, type_pb2 | 
|  | from google.protobuf import source_context_pb2, api_pb2 | 
|  |  | 
|  | HERE = os.path.dirname(os.path.abspath(__file__)) | 
|  | UPSTREAM = "https://server.codeium.com" | 
|  | AUTH_PATH = "/exa.auth_pb.AuthService/GetUserJwt" | 
|  | CHAT_PATH = "/exa.api_server_pb.ApiServerService/GetChatMessage" | 
|  | IDE_NAME = "windsurf" | 
|  | IDE_VERSION = "3.51.3" | 
|  | EXT_VERSION = "1.48.2" | 
|  | SESSION_PREFIX = "devin-session-token$" | 
|  | STOP_PATTERNS = ["<\|user\|>", "<\|bot\|>", "<\|context_request\|>", "<\|endoftext\|>", "<\|end_of_turn\|>"] | 
|  |  | 
|  | MODELS = [ | 
|  | "swe-2-max", "swe-2-high", "swe-2-medium", | 
|  | "swe-1-7", "swe-1-7-medium", "swe-1-7-lightning", "swe-1-7-lightning-medium", | 
|  | "swe-1-6", "swe-1-6-fast", | 
|  | ] | 
|  |  | 
|  | SRC_USER, SRC_SYSTEM, SRC_TOOL = 1, 2, 4 | 
|  | REQ_CASCADE, PLANNER_DEFAULT = 5, 1 | 
|  | CACHE_EPHEMERAL = 1 | 
|  | STOP_MAX_TOKENS = 3 | 
|  |  | 
|  | pool = descriptor_pool.DescriptorPool() | 
|  | for _m in (timestamp_pb2, duration_pb2, any_pb2, struct_pb2, descriptor_pb2, | 
|  | wrappers_pb2, empty_pb2, field_mask_pb2, type_pb2, source_context_pb2, api_pb2): | 
|  | _fd = descriptor_pb2.FileDescriptorProto() | 
|  | _fd.ParseFromString(_m.DESCRIPTOR.serialized_pb) | 
|  | try: | 
|  | pool.Add(_fd) | 
|  | except Exception: | 
|  | pass | 
|  | _fdps = {} | 
|  | for _p in glob.glob(os.path.join(HERE, "*.fdp")): | 
|  | _fd = descriptor_pb2.FileDescriptorProto() | 
|  | _fd.ParseFromString(open(_p, "rb").read()) | 
|  | _fdps[_fd.name] = _fd | 
|  | _added = set() | 
|  | for _ in range(40): | 
|  | for _name, _fd in _fdps.items(): | 
|  | if _name in _added: | 
|  | continue | 
|  | try: | 
|  | pool.Add(_fd) | 
|  | _added.add(_name) | 
|  | except Exception: | 
|  | pass | 
|  |  | 
|  |  | 
|  | def _msg(name): | 
|  | return message_factory.GetMessageClass(pool.FindMessageTypeByName(name)) | 
|  |  | 
|  |  | 
|  | GetUserJwtRequest = _msg("exa.auth_pb.GetUserJwtRequest") | 
|  | GetUserJwtResponse = _msg("exa.auth_pb.GetUserJwtResponse") | 
|  | GetChatMessageRequest = _msg("exa.api_server_pb.GetChatMessageRequest") | 
|  | GetChatMessageResponse = _msg("exa.api_server_pb.GetChatMessageResponse") | 
|  |  | 
|  |  | 
|  | def _load_key(): | 
|  | if os.environ.get("DEVIN_SHIM_API_KEY"): | 
|  | return os.environ["DEVIN_SHIM_API_KEY"] | 
|  | # Prefer the shim's own login (created via `XDG_DATA_HOME=~/devin-shim/data | 
|  | # devin auth login`); fall back to the Devin CLI's credential. | 
|  | for cred in (os.path.expanduser("~/devin-shim/data/devin/credentials.toml"), | 
|  | os.path.expanduser("~/.local/share/devin/credentials.toml")): | 
|  | if not os.path.exists(cred): | 
|  | continue | 
|  | for line in open(cred): | 
|  | if line.startswith("windsurf_api_key"): | 
|  | return line.split('"')[1] | 
|  | raise RuntimeError("no devin credential found") | 
|  |  | 
|  |  | 
|  | API_KEY = _load_key() | 
|  | if not API_KEY.startswith(SESSION_PREFIX): | 
|  | API_KEY = SESSION_PREFIX + API_KEY | 
|  |  | 
|  | _jwt_lock = threading.Lock() | 
|  | _jwt = {"token": None, "exp": 0.0, "base": None} | 
|  |  | 
|  |  | 
|  | def _metadata(jwt=""): | 
|  | return { | 
|  | "api_key": API_KEY, | 
|  | "user_jwt": jwt, | 
|  | "ide_name": IDE_NAME, | 
|  | "ide_version": IDE_VERSION, | 
|  | "extension_name": "windsurf", | 
|  | "extension_version": EXT_VERSION, | 
|  | "locale": "en", | 
|  | "session_id": str(uuid.uuid4()), | 
|  | "request_id": uuid.uuid4().int & (2**63 - 1), | 
|  | } | 
|  |  | 
|  |  | 
|  | def _jwt_expiry(token): | 
|  | try: | 
|  | import base64 | 
|  | payload = token.split(".")[1] | 
|  | payload += "=" * (-len(payload) % 4) | 
|  | return json.loads(base64.urlsafe_b64decode(payload)).get("exp", 0) | 
|  | except Exception: | 
|  | return 0.0 | 
|  |  | 
|  |  | 
|  | def get_jwt(force=False): | 
|  | with _jwt_lock: | 
|  | now = time.time() | 
|  | if not force and _jwt["token"] and _jwt["exp"] - 60 > now: | 
|  | return _jwt["token"], _jwt["base"] | 
|  | req = GetUserJwtRequest(metadata=_metadata()) | 
|  | r = requests.post(UPSTREAM + AUTH_PATH, data=req.SerializeToString(), | 
|  | headers={"content-type": "application/proto", | 
|  | "connect-protocol-version": "1"}, timeout=30) | 
|  | r.raise_for_status() | 
|  | resp = GetUserJwtResponse() | 
|  | try: | 
|  | resp.ParseFromString(r.content) | 
|  | except Exception: | 
|  | resp.ParseFromString(gzip.decompress(r.content)) | 
|  | if not resp.user_jwt: | 
|  | raise RuntimeError("GetUserJwt returned empty jwt") | 
|  | _jwt["token"] = resp.user_jwt | 
|  | _jwt["exp"] = _jwt_expiry(resp.user_jwt) or now + 3300 | 
|  | _jwt["base"] = resp.custom_api_server_url.strip() or None | 
|  | return _jwt["token"], _jwt["base"] | 
|  |  | 
|  |  | 
|  | def _text_of(content): | 
|  | if isinstance(content, str): | 
|  | return content | 
|  | out = [] | 
|  | for part in content or []: | 
|  | if part.get("type") == "text": | 
|  | out.append(part.get("text", "")) | 
|  | return "".join(out) | 
|  |  | 
|  |  | 
|  | # Cognition's input classifier rejects these Claude Code system-prompt blocks | 
|  | # (competitor identity, security-policy vocabulary, product marketing + URLs). | 
|  | # Rewrite them to neutral equivalents; behavior instructions are unchanged. | 
|  | _SYS_REWRITES = [ | 
|  | (re.compile(r"You are (?:a )?Claude[^.]*\."), "You are a coding agent."), | 
|  | (re.compile(r"IMPORTANT: Assist with authorized security testing.*?(?=\n\s*\n\|\Z)", re.S), | 
|  | "Assist with authorized security testing and defensive security work; refuse harmful or destructive requests."), | 
|  | (re.compile(r"\n?\s*-?\s*Claude Code is available[^\n]*"), ""), | 
|  | (re.compile(r"- For clear communication with the user the assistant MUST avoid using emojis\."), | 
|  | "- For clear communication with the user, avoid emojis."), | 
|  | ] | 
|  |  | 
|  |  | 
|  | def _scrub_system(text): | 
|  | for rx, rep in _SYS_REWRITES: | 
|  | text = rx.sub(rep, text) | 
|  | return text | 
|  |  | 
|  |  | 
|  | # TaskOutput's shipped description trips the same classifier in combination | 
|  | # (per-line fragments pass). It is deprecated upstream; send a short equivalent. | 
|  | _TOOL_DESC_REWRITES = { | 
|  | "TaskOutput": "Get the output of a running or completed background task (shell, agent, or remote session) by task_id.", | 
|  | } | 
|  |  | 
|  |  | 
|  | def _images_of(content): | 
|  | if isinstance(content, str) or not content: | 
|  | return [] | 
|  | out = [] | 
|  | for part in content or []: | 
|  | if part.get("type") == "image_url": | 
|  | url = part.get("image_url", {}).get("url", "") | 
|  | if url.startswith("data:"): | 
|  | mime, _, b64 = url[5:].partition(";base64,") | 
|  | out.append({"base64_data": b64, "mime_type": mime or "image/png"}) | 
|  | return out | 
|  |  | 
|  |  | 
|  | def build_request(body): | 
|  | system_parts, prompts = [], [] | 
|  | cascade_id = str(uuid.uuid4()) | 
|  | for i, m in enumerate(body.get("messages", [])): | 
|  | role = m.get("role") | 
|  | mid = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{cascade_id}\0{i}\0{role}")) | 
|  | if role in ("system", "developer"): | 
|  | system_parts.append(_scrub_system(_text_of(m.get("content")))) | 
|  | elif role == "user": | 
|  | prompts.append({"message_id": mid, "source": SRC_USER, | 
|  | "prompt": _text_of(m.get("content")), | 
|  | "images": _images_of(m.get("content"))}) | 
|  | elif role == "assistant": | 
|  | tcs = [{"id": tc.get("id", ""), "name": tc.get("function", {}).get("name", ""), | 
|  | "arguments_json": tc.get("function", {}).get("arguments", "")} | 
|  | for tc in m.get("tool_calls") or []] | 
|  | text = _text_of(m.get("content")) | 
|  | thinking = m.get("reasoning_content") or "" | 
|  | prompts.append({"message_id": mid, "source": SRC_SYSTEM, "prompt": text, | 
|  | "thinking": thinking, "tool_calls": tcs}) | 
|  | elif role == "tool": | 
|  | prompts.append({"message_id": mid, "source": SRC_TOOL, | 
|  | "tool_call_id": m.get("tool_call_id", ""), | 
|  | "prompt": _text_of(m.get("content")), | 
|  | "images": _images_of(m.get("content"))}) | 
|  | tools = [{"name": t["function"]["name"], | 
|  | "description": _TOOL_DESC_REWRITES.get(t["function"]["name"], | 
|  | t["function"].get("description", "")), | 
|  | "json_schema_string": json.dumps(t["function"].get("parameters") or {}), | 
|  | "strict": bool(t["function"].get("strict"))} | 
|  | for t in body.get("tools") or [] if t.get("type") == "function"] | 
|  | tc = body.get("tool_choice") | 
|  | tool_choice = {"option_name": "auto"} | 
|  | if isinstance(tc, str) and tc in ("auto", "required", "none"): | 
|  | tool_choice = {"option_name": tc} | 
|  | elif isinstance(tc, dict): | 
|  | fn = (tc.get("function") or {}).get("name") | 
|  | if fn: | 
|  | tool_choice = {"tool_name": fn} | 
|  | stops = list(STOP_PATTERNS) | 
|  | stop = body.get("stop") | 
|  | stops += [stop] if isinstance(stop, str) else list(stop or []) | 
|  | conf = {"num_completions": 1, "max_newlines": 200, "top_k": 50, | 
|  | "stop_patterns": stops, "fim_eot_prob_threshold": 1} | 
|  | conf["max_tokens"] = int(body.get("max_completion_tokens") or body.get("max_tokens") or 64000) | 
|  | if body.get("temperature") is not None: | 
|  | conf["temperature"] = conf["first_temperature"] = float(body["temperature"]) | 
|  | else: | 
|  | conf["temperature"] = conf["first_temperature"] = 0.4 | 
|  | if body.get("top_p") is not None: | 
|  | conf["top_p"] = float(body["top_p"]) | 
|  | else: | 
|  | conf["top_p"] = 1 | 
|  | model = body.get("model", "swe-2-max") | 
|  | if "/" in model: | 
|  | model = model.rsplit("/", 1)[-1] | 
|  | return GetChatMessageRequest( | 
|  | metadata=_metadata(get_jwt()[0]), | 
|  | prompt="\n\n".join(p for p in system_parts if p), | 
|  | chat_message_prompts=prompts, | 
|  | chat_model_uid=model, | 
|  | request_type=REQ_CASCADE, | 
|  | planner_mode=PLANNER_DEFAULT, | 
|  | tool_choice=tool_choice, | 
|  | system_prompt_cache_options={"type": CACHE_EPHEMERAL}, | 
|  | disable_parallel_tool_calls=False, | 
|  | cascade_id=cascade_id, | 
|  | execution_id=str(uuid.uuid4()), | 
|  | configuration=conf, | 
|  | tools=tools, | 
|  | ), model | 
|  |  | 
|  |  | 
|  | def chat_stream(req): | 
|  | """Yield (GetChatMessageResponse, None) per frame or (None, error_str) on trailer error.""" | 
|  | jwt, base = get_jwt() | 
|  | req.metadata.user_jwt = jwt | 
|  | body = req.SerializeToString() | 
|  | for attempt in range(2): | 
|  | gz = gzip.compress(body) | 
|  | frame = bytes([1]) + struct.pack(">I", len(gz)) + gz | 
|  | r = requests.post((base or UPSTREAM) + CHAT_PATH, data=frame, | 
|  | headers={"content-type": "application/connect+proto", | 
|  | "connect-protocol-version": "1", | 
|  | "connect-content-encoding": "gzip", | 
|  | "connect-accept-encoding": "gzip", | 
|  | "user-agent": "connect-go/1.18.1 (go1.26.3)"}, | 
|  | timeout=600, stream=True) | 
|  | if r.status_code == 200: | 
|  | break | 
|  | if r.status_code in (401, 403) and attempt == 0: | 
|  | jwt, base = get_jwt(force=True) | 
|  | req.metadata.user_jwt = jwt | 
|  | body = req.SerializeToString() | 
|  | continue | 
|  | yield None, f"upstream {r.status_code}: {r.text[:400]}" | 
|  | return | 
|  | buf = b"" | 
|  | for chunk in r.iter_content(65536): | 
|  | buf += chunk | 
|  | while len(buf) >= 5: | 
|  | flag = buf[0] | 
|  | ln = struct.unpack(">I", buf[1:5])[0] | 
|  | if len(buf) < 5 + ln: | 
|  | break | 
|  | payload = buf[5:5 + ln] | 
|  | buf = buf[5 + ln:] | 
|  | if flag & 2: | 
|  | trailer = gzip.decompress(payload) if flag & 1 else payload | 
|  | try: | 
|  | err = json.loads(trailer).get("error") or {} | 
|  | except Exception: | 
|  | err = {} | 
|  | if err.get("message"): | 
|  | yield None, f"{err.get('code', 'error')}: {err['message']}" | 
|  | continue | 
|  | raw = gzip.decompress(payload) if flag & 1 else payload | 
|  | msg = GetChatMessageResponse() | 
|  | msg.ParseFromString(raw) | 
|  | yield msg, None | 
|  |  | 
|  |  | 
|  | def openai_chunk(model, delta=None, finish=None, usage=None): | 
|  | ch = {"index": 0} | 
|  | if delta is not None: | 
|  | ch["delta"] = delta | 
|  | if finish: | 
|  | ch["finish_reason"] = finish | 
|  | out = {"id": "chatcmpl-devin", "object": "chat.completion.chunk", | 
|  | "created": int(time.time()), "model": model, "choices": [ch]} | 
|  | if usage: | 
|  | out["usage"] = usage | 
|  | return out | 
|  |  | 
|  |  | 
|  | def sse(obj): | 
|  | return f"data: {json.dumps(obj)}\n\n".encode() | 
|  |  | 
|  |  | 
|  | def run_chat(body, wfile): | 
|  | """Translate one OpenAI chat.completions call; returns (final_msg_dict, usage, error).""" | 
|  | try: | 
|  | req, model = build_request(body) | 
|  | except Exception as e: | 
|  | return None, None, f"request build: {e}" | 
|  | stream = bool(body.get("stream")) | 
|  | w = wfile if stream else None | 
|  | headers_sent = False | 
|  | # Cognition's input classifier denies borderline payloads nondeterministically | 
|  | # (same body observed pass/fail). Retry while nothing reached the client. | 
|  | for attempt in range(3): | 
|  | if attempt: | 
|  | req, model = build_request(body) | 
|  | if w and not headers_sent: | 
|  | w.write(b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n" | 
|  | b"cache-control: no-cache\r\nconnection: close\r\n\r\n") | 
|  | w.flush() | 
|  | w.write(sse(openai_chunk(model, delta={"role": "assistant"}))) | 
|  | w.flush() | 
|  | headers_sent = True | 
|  | text, thinking = [], [] | 
|  | tool_blocks = {}   # id -> {"name":..., "json":...} | 
|  | tool_order = [] | 
|  | usage = {} | 
|  | stop = 0 | 
|  | err = None | 
|  | emitted = False | 
|  | for msg, e in chat_stream(req): | 
|  | if e: | 
|  | err = e | 
|  | break | 
|  | if msg.delta_text: | 
|  | text.append(msg.delta_text) | 
|  | if w: | 
|  | emitted = True | 
|  | w.write(sse(openai_chunk(model, delta={"content": msg.delta_text}))) | 
|  | w.flush() | 
|  | if msg.delta_thinking: | 
|  | thinking.append(msg.delta_thinking) | 
|  | if w: | 
|  | emitted = True | 
|  | w.write(sse(openai_chunk(model, delta={"reasoning_content": msg.delta_thinking}))) | 
|  | w.flush() | 
|  | for tc in msg.delta_tool_calls: | 
|  | tid = tc.id or (tool_order[-1] if tool_order else "") | 
|  | if not tid: | 
|  | continue | 
|  | if tid not in tool_blocks: | 
|  | tool_blocks[tid] = {"name": tc.name, "json": ""} | 
|  | tool_order.append(tid) | 
|  | if w: | 
|  | emitted = True | 
|  | idx = tool_order.index(tid) | 
|  | w.write(sse(openai_chunk(model, delta={"tool_calls": [ | 
|  | {"index": idx, "id": tid, "type": "function", | 
|  | "function": {"name": tc.name, "arguments": ""}}]}))) | 
|  | w.flush() | 
|  | if tc.name: | 
|  | tool_blocks[tid]["name"] = tc.name | 
|  | if tc.arguments_json: | 
|  | prev = tool_blocks[tid]["json"] | 
|  | acc = tc.arguments_json if tc.arguments_json.startswith(prev) else prev + tc.arguments_json | 
|  | delta = acc[len(prev):] | 
|  | tool_blocks[tid]["json"] = acc | 
|  | if w and delta: | 
|  | emitted = True | 
|  | idx = tool_order.index(tid) | 
|  | w.write(sse(openai_chunk(model, delta={"tool_calls": [ | 
|  | {"index": idx, "function": {"arguments": delta}}]}))) | 
|  | w.flush() | 
|  | if msg.usage.input_tokens: | 
|  | usage = {"prompt_tokens": int(msg.usage.input_tokens), | 
|  | "completion_tokens": int(msg.usage.output_tokens), | 
|  | "total_tokens": int(msg.usage.input_tokens + msg.usage.output_tokens)} | 
|  | stop = msg.stop_reason | 
|  | if err and not emitted and attempt < 2 and "permission_denied" in err: | 
|  | print(f"upstream permission_denied, retrying (attempt {attempt + 2}/3)", flush=True) | 
|  | continue | 
|  | break | 
|  | if err: | 
|  | if w: | 
|  | w.write(sse({"error": {"message": err, "type": "upstream_error"}})) | 
|  | w.write(b"data: [DONE]\n\n") | 
|  | w.flush() | 
|  | return None, None, err | 
|  | finish = "tool_calls" if tool_blocks else "stop" | 
|  | if not tool_blocks and stop == STOP_MAX_TOKENS: | 
|  | finish = "length" | 
|  | if w: | 
|  | final = openai_chunk(model, delta={}, finish=finish, | 
|  | usage=usage if (body.get("stream_options") or {}).get("include_usage") else None) | 
|  | w.write(sse(final)) | 
|  | w.write(b"data: [DONE]\n\n") | 
|  | w.flush() | 
|  | return None, None, None | 
|  | msg_out = {"role": "assistant", "content": "".join(text)} | 
|  | if thinking: | 
|  | msg_out["reasoning_content"] = "".join(thinking) | 
|  | if tool_blocks: | 
|  | msg_out["tool_calls"] = [ | 
|  | {"id": tid, "type": "function", | 
|  | "function": {"name": b["name"], "arguments": b["json"]}} | 
|  | for tid, b in ((tid, tool_blocks[tid]) for tid in tool_order)] | 
|  | msg_out["content"] = msg_out["content"] or None | 
|  | resp = {"id": "chatcmpl-devin", "object": "chat.completion", | 
|  | "created": int(time.time()), "model": model, | 
|  | "choices": [{"index": 0, "message": msg_out, "finish_reason": finish}]} | 
|  | if usage: | 
|  | resp["usage"] = usage | 
|  | return resp, usage, None | 
|  |  | 
|  |  | 
|  | class Handler(BaseHTTPRequestHandler): | 
|  | protocol_version = "HTTP/1.1" | 
|  |  | 
|  | def log_message(self, fmt, *args): | 
|  | print(f"{self.address_string()} {fmt % args}", flush=True) | 
|  |  | 
|  | def _json(self, code, obj): | 
|  | data = json.dumps(obj).encode() | 
|  | self.send_response(code) | 
|  | self.send_header("content-type", "application/json") | 
|  | self.send_header("content-length", str(len(data))) | 
|  | self.end_headers() | 
|  | self.wfile.write(data) | 
|  |  | 
|  | def do_GET(self): | 
|  | if self.path.split("?")[0] in ("/v1/models", "/models"): | 
|  | self._json(200, {"object": "list", "data": [ | 
|  | {"id": m, "object": "model", "created": 0, "owned_by": "devin"} for m in MODELS]}) | 
|  | else: | 
|  | self._json(404, {"error": {"message": "not found", "type": "invalid_request_error"}}) | 
|  |  | 
|  | def do_POST(self): | 
|  | path = self.path.split("?")[0] | 
|  | if path not in ("/v1/chat/completions", "/chat/completions"): | 
|  | self._json(404, {"error": {"message": "not found", "type": "invalid_request_error"}}) | 
|  | return | 
|  | try: | 
|  | raw = self.rfile.read(int(self.headers.get("content-length", 0))) | 
|  | if os.environ.get("DEVIN_SHIM_DUMP"): | 
|  | with open(os.environ["DEVIN_SHIM_DUMP"] + f".{time.time_ns()}.json", "wb") as fh: | 
|  | fh.write(raw) | 
|  | body = json.loads(raw) | 
|  | except Exception as e: | 
|  | self._json(400, {"error": {"message": f"bad json: {e}", "type": "invalid_request_error"}}) | 
|  | return | 
|  | if body.get("stream"): | 
|  | # run_chat owns the raw socket from here | 
|  | resp, _, err = run_chat(body, self.wfile) | 
|  | if err: | 
|  | return | 
|  | return | 
|  | resp, _, err = run_chat(body, None) | 
|  | if err: | 
|  | self._json(502, {"error": {"message": err, "type": "upstream_error"}}) | 
|  | else: | 
|  | self._json(200, resp) | 
|  |  | 
|  |  | 
|  | if __name__ == "__main__": | 
|  | port = int(os.environ.get("DEVIN_SHIM_PORT", "8321")) | 
|  | print(f"devin-shim listening on 127.0.0.1:{port}", flush=True) | 
|  | ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever() |