635 lines
22 KiB
Python
Executable file
635 lines
22 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import http.client
|
|
import json
|
|
import os
|
|
import ssl
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
HOST = "127.0.0.1"
|
|
PORT = int(os.environ.get("IOP_CLAUDE_GATEWAY_PORT", "18443"))
|
|
|
|
SEULGIVIBE_HOST = os.environ.get("SEULGIVIBE_HOST", "seulgivibe.lgudax.cool")
|
|
SEULGIVIBE_PREFIX = os.environ.get("SEULGIVIBE_ANTHROPIC_PREFIX", "/anthropic")
|
|
SEULGIVIBE_KEY_HELPER = os.path.expanduser("~/.claude/anthropic_key.sh")
|
|
|
|
DEV_CORP_OPENAI_BASE_URL = os.environ.get(
|
|
"DEV_CORP_OPENAI_BASE_URL",
|
|
"http://digitalplatform.iop.ai.kr:18086/v1",
|
|
).rstrip("/")
|
|
DEV_CORP_KEY_HELPER = os.path.expanduser("~/.claude/dev-corp-openai-key.sh")
|
|
DEV_CORP_TOKEN_FILES = [
|
|
os.path.expanduser("~/.claude/dev-corp-openai-api-key"),
|
|
os.path.expanduser("~/.pi/agent/dev-corp-openai-api-key"),
|
|
]
|
|
DEV_CORP_MODEL = "claude-dev-corp-gemma4-26b"
|
|
DEV_CORP_UPSTREAM_MODEL = "gemma4:26b"
|
|
DEV_CORP_MIN_MAX_TOKENS = int(os.environ.get("DEV_CORP_UPSTREAM_MIN_MAX_TOKENS", "1024"))
|
|
|
|
SEULGIVIBE_ALIASES = {
|
|
"claude-seulgivibe-sonnet-4-5": "claude-sonnet-4-5-20250929",
|
|
"claude-seulgivibe-opus-4-8": "claude-opus-4-8",
|
|
"claude-seulgivibe-mythos-5": "claude-fable-5",
|
|
}
|
|
|
|
MODELS = [
|
|
{
|
|
"id": "claude-seulgivibe-sonnet-4-5",
|
|
"type": "model",
|
|
"display_name": "Seulgivibe Sonnet 4.5",
|
|
"description": "Seulgivibe Sonnet 4.5",
|
|
"created_at": "2026-05-13T00:00:00Z",
|
|
},
|
|
{
|
|
"id": "claude-seulgivibe-opus-4-8",
|
|
"type": "model",
|
|
"display_name": "Seulgivibe Opus 4.8",
|
|
"description": "Seulgivibe Opus 4.8",
|
|
"created_at": "2026-05-13T00:00:00Z",
|
|
},
|
|
{
|
|
"id": "claude-seulgivibe-mythos-5",
|
|
"type": "model",
|
|
"display_name": "Seulgivibe Fable 5",
|
|
"description": "Fable 5 · Most capable for your hardest and longest-running tasks · $10/$50 per Mtok",
|
|
"created_at": "2026-05-13T00:00:00Z",
|
|
},
|
|
{
|
|
"id": DEV_CORP_MODEL,
|
|
"type": "model",
|
|
"display_name": "dev-corp gemma4:26b",
|
|
"description": "dev-corp gemma4:26b via OpenAI-compatible gateway",
|
|
"created_at": "2026-07-10T00:00:00Z",
|
|
},
|
|
]
|
|
|
|
|
|
def is_record(value):
|
|
return isinstance(value, dict)
|
|
|
|
|
|
def content_to_text(content):
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts = []
|
|
for item in content:
|
|
if isinstance(item, str):
|
|
parts.append(item)
|
|
elif is_record(item):
|
|
item_type = item.get("type")
|
|
if item_type in ("text", "input_text", "output_text"):
|
|
text = item.get("text")
|
|
if isinstance(text, str):
|
|
parts.append(text)
|
|
elif item_type == "tool_result":
|
|
text = content_to_text(item.get("content"))
|
|
if text:
|
|
parts.append(text)
|
|
return "\n".join(parts)
|
|
return ""
|
|
|
|
|
|
def normalize_system(system_value):
|
|
if isinstance(system_value, str):
|
|
return system_value
|
|
if isinstance(system_value, list):
|
|
return content_to_text(system_value)
|
|
return ""
|
|
|
|
|
|
def normalize_system_messages(payload):
|
|
messages = payload.get("messages")
|
|
if not isinstance(messages, list):
|
|
return
|
|
|
|
system_parts = []
|
|
regular_messages = []
|
|
for message in messages:
|
|
if is_record(message) and message.get("role") == "system":
|
|
text = content_to_text(message.get("content"))
|
|
if text:
|
|
system_parts.append(text)
|
|
else:
|
|
regular_messages.append(message)
|
|
|
|
if not system_parts:
|
|
return
|
|
|
|
existing = normalize_system(payload.get("system"))
|
|
payload["system"] = "\n\n".join(part for part in [existing, *system_parts] if part)
|
|
payload["messages"] = regular_messages
|
|
|
|
|
|
def read_helper(path):
|
|
if not os.path.exists(path):
|
|
return ""
|
|
try:
|
|
result = subprocess.run(
|
|
[path],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
except Exception:
|
|
return ""
|
|
return result.stdout.strip()
|
|
|
|
|
|
def read_token_file(path):
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
return handle.read().strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def inbound_token(headers):
|
|
for key in ("authorization", "x-api-key", "api-key"):
|
|
value = headers.get(key)
|
|
if not value:
|
|
continue
|
|
value = value.strip()
|
|
if value.lower().startswith("bearer "):
|
|
return value[7:].strip()
|
|
return value
|
|
return ""
|
|
|
|
|
|
def seulgivibe_token(headers):
|
|
return inbound_token(headers) or read_helper(SEULGIVIBE_KEY_HELPER)
|
|
|
|
|
|
def dev_corp_token(headers):
|
|
token = read_helper(DEV_CORP_KEY_HELPER)
|
|
if token:
|
|
return token
|
|
for path in DEV_CORP_TOKEN_FILES:
|
|
token = read_token_file(path)
|
|
if token:
|
|
return token
|
|
return inbound_token(headers) or read_helper(SEULGIVIBE_KEY_HELPER)
|
|
|
|
|
|
def seulgivibe_path(raw_path):
|
|
if raw_path.startswith(f"{SEULGIVIBE_PREFIX}/"):
|
|
return raw_path
|
|
if raw_path.startswith("/v1/"):
|
|
return f"{SEULGIVIBE_PREFIX}{raw_path}"
|
|
return raw_path
|
|
|
|
|
|
def call_seulgivibe(method, raw_path, payload, inbound_headers):
|
|
outbound_payload = dict(payload)
|
|
model = outbound_payload.get("model")
|
|
if model in SEULGIVIBE_ALIASES:
|
|
normalize_system_messages(outbound_payload)
|
|
outbound_payload["model"] = SEULGIVIBE_ALIASES[model]
|
|
|
|
body = json.dumps(outbound_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
headers = {
|
|
"content-type": "application/json",
|
|
"accept": inbound_headers.get("accept", "application/json, text/event-stream"),
|
|
"content-length": str(len(body)),
|
|
}
|
|
for key in ("anthropic-version", "anthropic-beta"):
|
|
if inbound_headers.get(key):
|
|
headers[key] = inbound_headers[key]
|
|
token = seulgivibe_token(inbound_headers)
|
|
if token:
|
|
headers["authorization"] = f"Bearer {token}"
|
|
|
|
context = ssl._create_unverified_context()
|
|
conn = http.client.HTTPSConnection(SEULGIVIBE_HOST, 443, context=context, timeout=300)
|
|
try:
|
|
conn.request(method, seulgivibe_path(raw_path), body=body, headers=headers)
|
|
response = conn.getresponse()
|
|
return response.status, dict(response.getheaders()), response.read()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def unwrap_json_string_body(body):
|
|
try:
|
|
value = json.loads(body.decode("utf-8"))
|
|
except Exception:
|
|
return body
|
|
if not isinstance(value, str):
|
|
return body
|
|
stripped = value.strip()
|
|
if not stripped.startswith(("{", "[")):
|
|
return body
|
|
return stripped.encode("utf-8")
|
|
|
|
|
|
def anthropic_messages_to_openai(payload):
|
|
messages = []
|
|
system_text = normalize_system(payload.get("system"))
|
|
if system_text:
|
|
messages.append({"role": "system", "content": system_text})
|
|
|
|
tool_names = {}
|
|
for message in payload.get("messages") or []:
|
|
if not is_record(message):
|
|
continue
|
|
role = message.get("role")
|
|
content = message.get("content")
|
|
|
|
if role == "assistant":
|
|
text_parts = []
|
|
tool_calls = []
|
|
blocks = content if isinstance(content, list) else [{"type": "text", "text": content_to_text(content)}]
|
|
for block in blocks:
|
|
if not is_record(block):
|
|
continue
|
|
block_type = block.get("type")
|
|
if block_type == "text":
|
|
text = block.get("text")
|
|
if isinstance(text, str) and text:
|
|
text_parts.append(text)
|
|
elif block_type == "tool_use":
|
|
tool_id = str(block.get("id") or f"call_{uuid.uuid4().hex[:16]}")
|
|
name = str(block.get("name") or "")
|
|
tool_names[tool_id] = name
|
|
tool_calls.append(
|
|
{
|
|
"id": tool_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"arguments": json.dumps(block.get("input") or {}, ensure_ascii=False),
|
|
},
|
|
}
|
|
)
|
|
openai_message = {"role": "assistant", "content": "\n".join(text_parts) or None}
|
|
if tool_calls:
|
|
openai_message["tool_calls"] = tool_calls
|
|
messages.append(openai_message)
|
|
continue
|
|
|
|
if role == "user":
|
|
text_parts = []
|
|
blocks = content if isinstance(content, list) else [{"type": "text", "text": content_to_text(content)}]
|
|
for block in blocks:
|
|
if is_record(block) and block.get("type") == "tool_result":
|
|
tool_call_id = str(block.get("tool_use_id") or "")
|
|
tool_message = {
|
|
"role": "tool",
|
|
"tool_call_id": tool_call_id,
|
|
"content": content_to_text(block.get("content")),
|
|
}
|
|
name = tool_names.get(tool_call_id)
|
|
if name:
|
|
tool_message["name"] = name
|
|
messages.append(tool_message)
|
|
else:
|
|
text = content_to_text([block])
|
|
if text:
|
|
text_parts.append(text)
|
|
if text_parts:
|
|
messages.append({"role": "user", "content": "\n".join(text_parts)})
|
|
continue
|
|
|
|
if role in ("system", "developer"):
|
|
text = content_to_text(content)
|
|
if text:
|
|
messages.append({"role": "system", "content": text})
|
|
|
|
return messages
|
|
|
|
|
|
def anthropic_tools_to_openai(payload):
|
|
tools = []
|
|
for tool in payload.get("tools") or []:
|
|
if not is_record(tool):
|
|
continue
|
|
name = tool.get("name")
|
|
if not isinstance(name, str) or not name:
|
|
continue
|
|
parameters = tool.get("input_schema")
|
|
if not is_record(parameters):
|
|
parameters = {"type": "object", "properties": {}}
|
|
tools.append(
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"description": tool.get("description") or "",
|
|
"parameters": parameters,
|
|
},
|
|
}
|
|
)
|
|
return tools
|
|
|
|
|
|
def build_openai_payload(payload):
|
|
body = {
|
|
"model": DEV_CORP_UPSTREAM_MODEL,
|
|
"messages": anthropic_messages_to_openai(payload),
|
|
"temperature": payload.get("temperature", 0.2),
|
|
"top_p": payload.get("top_p", 0.95),
|
|
"stream": False,
|
|
}
|
|
max_tokens = payload.get("max_tokens")
|
|
if isinstance(max_tokens, int) and max_tokens > 0:
|
|
body["max_tokens"] = max(max_tokens, DEV_CORP_MIN_MAX_TOKENS)
|
|
else:
|
|
body["max_tokens"] = DEV_CORP_MIN_MAX_TOKENS
|
|
tools = anthropic_tools_to_openai(payload)
|
|
if tools:
|
|
body["tools"] = tools
|
|
return body
|
|
|
|
|
|
def parse_openai_url(path):
|
|
parsed = urlsplit(DEV_CORP_OPENAI_BASE_URL)
|
|
scheme = parsed.scheme or "http"
|
|
host = parsed.hostname or ""
|
|
port = parsed.port
|
|
prefix = parsed.path.rstrip("/")
|
|
return scheme, host, port, f"{prefix}{path}"
|
|
|
|
|
|
def call_dev_corp(openai_payload, inbound_headers):
|
|
scheme, host, port, path = parse_openai_url("/chat/completions")
|
|
body = json.dumps(openai_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
headers = {
|
|
"content-type": "application/json",
|
|
"accept": "application/json",
|
|
"content-length": str(len(body)),
|
|
}
|
|
token = dev_corp_token(inbound_headers)
|
|
if token:
|
|
headers["authorization"] = f"Bearer {token}"
|
|
|
|
if scheme == "https":
|
|
context = ssl._create_unverified_context()
|
|
conn = http.client.HTTPSConnection(host, port or 443, context=context, timeout=300)
|
|
else:
|
|
conn = http.client.HTTPConnection(host, port or 80, timeout=300)
|
|
try:
|
|
conn.request("POST", path, body=body, headers=headers)
|
|
response = conn.getresponse()
|
|
return response.status, dict(response.getheaders()), response.read()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def parse_tool_arguments(raw):
|
|
if not isinstance(raw, str) or raw == "":
|
|
return {}
|
|
try:
|
|
value = json.loads(raw)
|
|
except Exception:
|
|
return {"raw": raw}
|
|
return value if is_record(value) else {"value": value}
|
|
|
|
|
|
def openai_to_anthropic(openai_payload, requested_model):
|
|
choice = (openai_payload.get("choices") or [{}])[0]
|
|
message = choice.get("message") or {}
|
|
finish_reason = choice.get("finish_reason")
|
|
content = []
|
|
|
|
text = message.get("content")
|
|
if isinstance(text, str) and text:
|
|
content.append({"type": "text", "text": text})
|
|
|
|
tool_calls = message.get("tool_calls") or []
|
|
for tool_call in tool_calls:
|
|
if not is_record(tool_call):
|
|
continue
|
|
function = tool_call.get("function") or {}
|
|
content.append(
|
|
{
|
|
"type": "tool_use",
|
|
"id": str(tool_call.get("id") or f"call_{uuid.uuid4().hex[:16]}"),
|
|
"name": str(function.get("name") or ""),
|
|
"input": parse_tool_arguments(function.get("arguments")),
|
|
}
|
|
)
|
|
|
|
usage = openai_payload.get("usage") or {}
|
|
if tool_calls:
|
|
stop_reason = "tool_use"
|
|
elif finish_reason == "length":
|
|
stop_reason = "max_tokens"
|
|
else:
|
|
stop_reason = "end_turn"
|
|
|
|
return {
|
|
"id": openai_payload.get("id") or f"msg_{uuid.uuid4().hex}",
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": requested_model or DEV_CORP_MODEL,
|
|
"content": content or [{"type": "text", "text": ""}],
|
|
"stop_reason": stop_reason,
|
|
"stop_sequence": None,
|
|
"usage": {
|
|
"input_tokens": int(usage.get("prompt_tokens") or 0),
|
|
"output_tokens": int(usage.get("completion_tokens") or 0),
|
|
},
|
|
}
|
|
|
|
|
|
def estimate_tokens(payload):
|
|
text = normalize_system(payload.get("system"))
|
|
for message in payload.get("messages") or []:
|
|
if is_record(message):
|
|
text += "\n" + content_to_text(message.get("content"))
|
|
return max(1, len(text) // 4)
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, *_):
|
|
return
|
|
|
|
def read_json(self):
|
|
length = int(self.headers.get("content-length", "0") or "0")
|
|
body = self.rfile.read(length) if length > 0 else b"{}"
|
|
return json.loads(body.decode("utf-8") or "{}")
|
|
|
|
def write_json(self, status, payload):
|
|
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("content-type", "application/json")
|
|
self.send_header("content-length", str(len(body)))
|
|
self.send_header("connection", "close")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
self.close_connection = True
|
|
|
|
def write_raw(self, status, headers, body):
|
|
self.send_response(status)
|
|
for key, value in headers.items():
|
|
lower_key = key.lower()
|
|
if lower_key in ("connection", "transfer-encoding", "content-length"):
|
|
continue
|
|
self.send_header(key, value)
|
|
self.send_header("content-length", str(len(body)))
|
|
self.send_header("connection", "close")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
self.close_connection = True
|
|
|
|
def write_sse(self, event, payload):
|
|
self.wfile.write(f"event: {event}\n".encode("utf-8"))
|
|
self.wfile.write(b"data: ")
|
|
self.wfile.write(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
|
|
self.wfile.write(b"\n\n")
|
|
self.wfile.flush()
|
|
|
|
def write_anthropic_stream(self, message):
|
|
self.send_response(200)
|
|
self.send_header("content-type", "text/event-stream")
|
|
self.send_header("cache-control", "no-cache")
|
|
self.send_header("connection", "close")
|
|
self.end_headers()
|
|
|
|
self.write_sse(
|
|
"message_start",
|
|
{
|
|
"type": "message_start",
|
|
"message": {
|
|
"id": message["id"],
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": message["model"],
|
|
"content": [],
|
|
"stop_reason": None,
|
|
"stop_sequence": None,
|
|
"usage": {"input_tokens": message["usage"]["input_tokens"], "output_tokens": 0},
|
|
},
|
|
},
|
|
)
|
|
for index, block in enumerate(message["content"]):
|
|
if block.get("type") == "tool_use":
|
|
start_block = {
|
|
"type": "tool_use",
|
|
"id": block.get("id"),
|
|
"name": block.get("name"),
|
|
"input": {},
|
|
}
|
|
self.write_sse(
|
|
"content_block_start",
|
|
{"type": "content_block_start", "index": index, "content_block": start_block},
|
|
)
|
|
partial_json = json.dumps(block.get("input") or {}, ensure_ascii=False, separators=(",", ":"))
|
|
if partial_json:
|
|
self.write_sse(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {"type": "input_json_delta", "partial_json": partial_json},
|
|
},
|
|
)
|
|
else:
|
|
self.write_sse(
|
|
"content_block_start",
|
|
{"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}},
|
|
)
|
|
text = block.get("text") or ""
|
|
if text:
|
|
self.write_sse(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {"type": "text_delta", "text": text},
|
|
},
|
|
)
|
|
self.write_sse("content_block_stop", {"type": "content_block_stop", "index": index})
|
|
|
|
self.write_sse(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": message["stop_reason"], "stop_sequence": None},
|
|
"usage": {"output_tokens": message["usage"]["output_tokens"]},
|
|
},
|
|
)
|
|
self.write_sse("message_stop", {"type": "message_stop"})
|
|
self.close_connection = True
|
|
|
|
def do_GET(self):
|
|
path = urlsplit(self.path).path
|
|
if path == "/health":
|
|
self.write_json(200, {"ok": True, "models": [model["id"] for model in MODELS]})
|
|
return
|
|
if path in ("/v1/models", "/anthropic/v1/models"):
|
|
self.write_json(200, {"data": MODELS, "first_id": MODELS[0]["id"], "last_id": MODELS[-1]["id"], "has_more": False})
|
|
return
|
|
self.write_json(404, {"type": "error", "error": {"type": "not_found_error", "message": path}})
|
|
|
|
def do_POST(self):
|
|
path = urlsplit(self.path).path
|
|
is_count_tokens = path in ("/v1/messages/count_tokens", "/anthropic/v1/messages/count_tokens")
|
|
is_messages = path in ("/v1/messages", "/anthropic/v1/messages")
|
|
if not is_count_tokens and not is_messages:
|
|
self.write_json(404, {"type": "error", "error": {"type": "not_found_error", "message": path}})
|
|
return
|
|
|
|
try:
|
|
payload = self.read_json()
|
|
except Exception as exc:
|
|
self.write_json(400, {"type": "error", "error": {"type": "invalid_request_error", "message": str(exc)}})
|
|
return
|
|
|
|
headers = {key.lower(): value for key, value in self.headers.items()}
|
|
if payload.get("model") == DEV_CORP_MODEL:
|
|
if is_count_tokens:
|
|
self.write_json(200, {"input_tokens": estimate_tokens(payload)})
|
|
return
|
|
openai_payload = build_openai_payload(payload)
|
|
status, _, body = call_dev_corp(openai_payload, headers)
|
|
if status < 200 or status >= 300:
|
|
try:
|
|
detail = json.loads(body.decode("utf-8"))
|
|
except Exception:
|
|
detail = body.decode("utf-8", "replace")
|
|
self.write_json(
|
|
502,
|
|
{
|
|
"type": "error",
|
|
"error": {
|
|
"type": "upstream_error",
|
|
"message": f"dev-corp upstream returned HTTP {status}",
|
|
"detail": detail,
|
|
},
|
|
},
|
|
)
|
|
return
|
|
try:
|
|
openai_response = json.loads(body.decode("utf-8"))
|
|
except Exception as exc:
|
|
self.write_json(502, {"type": "error", "error": {"type": "upstream_parse_error", "message": str(exc)}})
|
|
return
|
|
message = openai_to_anthropic(openai_response, payload.get("model"))
|
|
if payload.get("stream") is True:
|
|
self.write_anthropic_stream(message)
|
|
else:
|
|
self.write_json(200, message)
|
|
return
|
|
|
|
status, response_headers, body = call_seulgivibe(self.command, path, payload, headers)
|
|
if 200 <= status < 300:
|
|
body = unwrap_json_string_body(body)
|
|
self.write_raw(status, response_headers, body)
|
|
|
|
|
|
class Server(ThreadingHTTPServer):
|
|
allow_reuse_address = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"iop claude gateway listening on http://{HOST}:{PORT} at {time.strftime('%Y-%m-%d %H:%M:%S')}", flush=True)
|
|
Server((HOST, PORT), Handler).serve_forever()
|