#!/usr/bin/env python3 """ ollama-proxy.py — Ollama + Anthropic API proxy backed by llama.cpp's OpenAI server. Fakes two API surfaces on one port (default 11434): * Ollama: /api/version, /api/tags, /api/ps, /api/show, /api/chat, /api/generate, /api/embed, /api/embeddings, /api/pull, /api/push, /api/create, /api/copy, /api/delete * Anthropic: /v1/messages, /v1/messages/count_tokens (what `ollama launch claude` and Claude Code actually speak) and forwards to a llama.cpp `llama-server` OpenAI-compatible backend (default http://127.0.0.1:8080/v1). Pure stdlib — no third-party deps. Requires Python 3.8+. """ import json import sys import time import uuid import urllib.request import urllib.error import http.server import threading import re import subprocess # ---------------------------------------------------------------------------- # GPU Power & Clock Governor (Automatic idle power reduction for CMP 50HX) # ---------------------------------------------------------------------------- class GPUGovernor: """Manages GPU clocks dynamically: drops CMP cards to low-power state when idle (saves ~90W), and instantly unconstrains to full boost clocks during inference.""" def __init__(self, idle_timeout=30): self.idle_timeout = idle_timeout self.last_active = time.time() self.is_low_power = False self.lock = threading.Lock() self._init_limits() t = threading.Thread(target=self._monitor_loop, daemon=True) t.start() def _run_cmd(self, *args): try: subprocess.run(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) except Exception: pass def _init_limits(self): # Ensure persistence mode and 150W safe power cap self._run_cmd("nvidia-smi", "-pm", "1") self._run_cmd("nvidia-smi", "-i", "1,2", "-pl", "150") def wake(self): with self.lock: self.last_active = time.time() if self.is_low_power: self._run_cmd("nvidia-smi", "-rgc") self.is_low_power = False def touch(self): with self.lock: self.last_active = time.time() def _monitor_loop(self): while True: time.sleep(5) with self.lock: if not self.is_low_power and (time.time() - self.last_active) > self.idle_timeout: # Drop CMP cards to low power idle clock self._run_cmd("nvidia-smi", "-i", "1,2", "-lgc", "600,600") self.is_low_power = True GOVERNOR = GPUGovernor(idle_timeout=30) # ---------------------------------------------------------------------------- # Config # ---------------------------------------------------------------------------- HOST = "0.0.0.0" PORT = 11434 BACKEND = "http://127.0.0.1:8080/v1" # llama.cpp OpenAI endpoint MODEL_NAME = "Qwen3.8-Uncensored" # primary name BACKEND_MODEL = MODEL_NAME # what we send llama.cpp VERSION = "0.5.4" # fake ollama version CTX_SIZE = 262144 MAX_OUTPUT = 131072 KEEP_ALIVE = 300 MODEL_TAGS = [ "Qwen3.8-Uncensored:latest", "Qwen3.8-Uncensored", "qwen3.8-uncensored:latest", "qwen3.8-uncensored", "qwen3.8:latest", "qwen3.8", "qwen:latest", "qwen", "qwen:fast", "qwen:nothink", "qwen:think", "qwen:reasoning", "Qwen3.8-27B-Uncensored:latest", "Qwen3.8-27B-Uncensored", "qwen3.8-27b:latest", "qwen3.8-27b", ] # Set to False to hide reasoning_content / thinking blocks entirely. EMIT_THINKING = True # ---------------------------------------------------------------------------- # Small helpers # ---------------------------------------------------------------------------- def now_iso(): return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + ".000000Z" def now_ns(): return time.time_ns() def est_tokens(text): # rough heuristic; used for count_tokens / usage when backend omits it if not text: return 0 return max(1, (len(text) + 3) // 4) def _post(path, payload, stream=False): GOVERNOR.wake() req = urllib.request.Request( BACKEND + path, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, method="POST", ) return urllib.request.urlopen(req, timeout=None) def _backend_chat(messages, **kw): """Call llama.cpp /chat/completions, return parsed JSON (non-streaming).""" body = {"model": BACKEND_MODEL, "messages": messages} body.update(kw) body.setdefault("stream", False) with _post("/chat/completions", body) as r: return json.loads(r.read().decode()) def _backend_stream(messages, **kw): """Yield raw SSE data lines from llama.cpp streaming chat.""" body = { "model": BACKEND_MODEL, "messages": messages, "stream": True, "stream_options": {"include_usage": True}, } body.update(kw) with _post("/chat/completions", body) as r: for raw in r: GOVERNOR.touch() line = raw.decode(errors="replace").strip() if not line: continue if line.startswith("data:"): yield line[len("data:"):].strip() else: yield line def _format_image_url(img_b64): s = str(img_b64).strip() if s.startswith("data:"): return s if s.startswith("iVBOR"): mtype = "image/png" elif s.startswith("/9j/"): mtype = "image/jpeg" elif s.startswith("R0lGOD"): mtype = "image/gif" elif s.startswith("UklGR"): mtype = "image/webp" else: mtype = "image/jpeg" return f"data:{mtype};base64,{s}" def _openai_msg_from_ollama_message(msg): """Convert an ollama chat message to an OpenAI-format message.""" role = msg.get("role", "user") content = msg.get("content", "") if role == "assistant" and "reasoning_content" in msg: # best-effort: fold prior reasoning in as plain content so context is kept rc = msg.get("reasoning_content") or "" content = content or rc images = msg.get("images", []) if images: content_list = [] if content: content_list.append({"type": "text", "text": content}) for img_b64 in images: content_list.append({"type": "image_url", "image_url": {"url": _format_image_url(img_b64)}}) out = {"role": role, "content": content_list} elif isinstance(content, list): out = {"role": role, "content": content} else: out = {"role": role, "content": content} # ollama tool_calls -> OpenAI tool_calls tool_calls = msg.get("tool_calls") if tool_calls: oc = [] for tc in tool_calls: fn = tc.get("function", {}) args = fn.get("arguments", {}) oc.append({ "id": tc.get("id") or "call_" + uuid.uuid4().hex[:16], "type": "function", "function": { "name": fn.get("name", ""), "arguments": json.dumps(args) if isinstance(args, dict) else str(args), }, }) out["tool_calls"] = oc return out def _ollama_tools_to_openai(tools): """ollama tools (OpenAI-shaped) -> OpenAI tools, passthrough with cleanup.""" if not tools: return None out = [] for t in tools: fn = t.get("function", t) out.append({ "type": "function", "function": { "name": fn.get("name", ""), "description": fn.get("description", ""), "parameters": fn.get("parameters") or fn.get("input_schema") or {}, }, }) return out def _openai_tool_calls_to_ollama(tool_calls): """OpenAI tool_calls -> ollama message.tool_calls.""" if not tool_calls: return None out = [] for tc in tool_calls: fn = tc.get("function", {}) try: args = json.loads(fn.get("arguments", "{}")) except json.JSONDecodeError: args = {} out.append({ "function": { "name": fn.get("name", ""), "arguments": args, }, }) return out def _ollama_error(code, msg): body = json.dumps({"error": msg}).encode() return body, code, "application/json" # ---------------------------------------------------------------------------- # Ollama API # ---------------------------------------------------------------------------- def ollama_version(): return {"version": VERSION} def _model_entry(tag): return { "name": tag, "model": tag, "modified_at": now_iso(), "size": 17923393664, "digest": "sha256:" + "0" * 64, "details": { "format": "gguf", "family": "qwen35", "families": ["qwen35", "qwen2"], "parameter_size": "27.0B", "quantization_level": "Q4_K_P", "context_length": CTX_SIZE, }, "capabilities": ["completion", "chat", "tools", "thinking"], } def ollama_tags(): return {"models": [_model_entry(tag) for tag in MODEL_TAGS]} def ollama_ps(): return { "models": [ { "name": "Qwen3.8-Uncensored:latest", "model": "Qwen3.8-Uncensored:latest", "size": 17923393664, "digest": "sha256:" + "0" * 64, "details": { "format": "gguf", "family": "qwen35", "families": ["qwen35", "qwen2"], "parameter_size": "27.0B", "quantization_level": "Q4_K_P", "context_length": CTX_SIZE, }, "expires_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(time.time() + 86400)) + ".000000Z", "size_vram": 17923393664, "processors": 3, "context_length": CTX_SIZE, } ] } def ollama_show(payload=None): req_model = (payload.get("name") or payload.get("model") or MODEL_NAME) if payload else MODEL_NAME return { "license": "Apache-2.0", "modelfile": f"FROM {req_model}\nPARAMETER temperature 0.7\nPARAMETER top_p 0.95\nPARAMETER num_ctx {CTX_SIZE}", "parameters": f"temperature 0.7\ntop_p 0.95\nnum_ctx {CTX_SIZE}", "template": "{{ if .System }}<|im_start|>system\n{{ .System }}<|im_end|>\n{{ end }}{{ if .Prompt }}<|im_start|>user\n{{ .Prompt }}<|im_end|>\n{{ end }}<|im_start|>assistant\n{{ .Response }}<|im_end|>", "system": "", "details": { "format": "gguf", "family": "qwen35", "families": ["qwen35", "qwen2"], "parameter_size": "27.0B", "quantization_level": "Q4_K_P", "context_length": CTX_SIZE, }, "model_info": { "general.architecture": "qwen35", "general.name": req_model, "general.parameter_count": 27320697856, "general.quantization_version": 2, "qwen35.context_length": CTX_SIZE, "qwen35.attention.head_count": 32, "qwen35.attention.head_count_kv": 4, "qwen35.max_output_tokens": MAX_OUTPUT, }, "capabilities": ["completion", "chat", "tools", "thinking"], } def ollama_status(): return {"cloud": {"disabled": True, "source": ""}} def ollama_recommendations(): return { "recommendations": [ { "model": tag, "description": "Qwen3.8-27B Uncensored (local, 256K context)", "context_length": CTX_SIZE, "max_output_tokens": MAX_OUTPUT, } for tag in MODEL_TAGS[:2] ] } def _map_options(opts, kw): if not opts or not isinstance(opts, dict): return if "temperature" in opts and opts["temperature"] is not None: kw["temperature"] = float(opts["temperature"]) if "top_p" in opts and opts["top_p"] is not None: kw["top_p"] = float(opts["top_p"]) if "top_k" in opts and opts["top_k"] is not None: kw["top_k"] = int(opts["top_k"]) if "min_p" in opts and opts["min_p"] is not None: kw["min_p"] = float(opts["min_p"]) if "num_predict" in opts and opts["num_predict"] is not None: kw["max_tokens"] = int(opts["num_predict"]) elif "max_tokens" in opts and opts["max_tokens"] is not None: kw["max_tokens"] = int(opts["max_tokens"]) if "stop" in opts and opts["stop"] is not None: kw["stop"] = opts["stop"] if "seed" in opts and opts["seed"] is not None: kw["seed"] = int(opts["seed"]) if "repeat_penalty" in opts and opts["repeat_penalty"] is not None: kw["repeat_penalty"] = float(opts["repeat_penalty"]) if "presence_penalty" in opts and opts["presence_penalty"] is not None: kw["presence_penalty"] = float(opts["presence_penalty"]) if "frequency_penalty" in opts and opts["frequency_penalty"] is not None: kw["frequency_penalty"] = float(opts["frequency_penalty"]) if "mirostat" in opts and opts["mirostat"] is not None: kw["mirostat"] = int(opts["mirostat"]) if "mirostat_tau" in opts and opts["mirostat_tau"] is not None: kw["mirostat_tau"] = float(opts["mirostat_tau"]) if "mirostat_eta" in opts and opts["mirostat_eta"] is not None: kw["mirostat_eta"] = float(opts["mirostat_eta"]) def _extract_thinking_params(payload): """Extract and resolve thinking/reasoning parameters across Ollama, OpenAI, and Anthropic formats.""" opts = payload.get("options", {}) or {} model_req = str(payload.get("model") or payload.get("name") or "").lower() enable_thinking = None reasoning_effort = None reasoning_budget = None # Check model alias for explicit thinking mode if any(k in model_req for k in [":fast", ":nothink", ":no-think", ":direct"]): enable_thinking = False elif any(k in model_req for k in [":think", ":reasoning", ":deep"]): enable_thinking = True # Check top-level payload parameters if "think" in payload and payload["think"] is not None: enable_thinking = bool(payload["think"]) elif "enable_thinking" in payload and payload["enable_thinking"] is not None: enable_thinking = bool(payload["enable_thinking"]) elif "thinking" in payload and payload["thinking"] is not None: val = payload["thinking"] if isinstance(val, bool): enable_thinking = val elif isinstance(val, dict): if val.get("type") in ("disabled", "off", "none"): enable_thinking = False elif val.get("type") in ("enabled", "on"): enable_thinking = True if "budget_tokens" in val and val["budget_tokens"] is not None: reasoning_budget = int(val["budget_tokens"]) # Check options dictionary (OpenWebUI / Ollama options) if "enable_thinking" in opts and opts["enable_thinking"] is not None: enable_thinking = bool(opts["enable_thinking"]) elif "thinking" in opts and opts["thinking"] is not None: val = opts["thinking"] if isinstance(val, bool): enable_thinking = val elif isinstance(val, dict): if val.get("type") in ("disabled", "off", "none"): enable_thinking = False elif val.get("type") in ("enabled", "on"): enable_thinking = True if "budget_tokens" in val and val["budget_tokens"] is not None: reasoning_budget = int(val["budget_tokens"]) elif "num_thinking" in opts and opts["num_thinking"] is not None: if int(opts["num_thinking"]) == 0: enable_thinking = False else: reasoning_budget = int(opts["num_thinking"]) elif "reasoning_budget" in opts and opts["reasoning_budget"] is not None: reasoning_budget = int(opts["reasoning_budget"]) # Check reasoning effort ('low', 'medium', 'high', 'xhigh', 'none', 'off') effort_raw = payload.get("reasoning_effort") or opts.get("reasoning_effort") if effort_raw is not None: effort_str = str(effort_raw).lower().strip() if effort_str in ("none", "off", "0", "false", "disabled", "no"): enable_thinking = False reasoning_effort = "low" elif effort_str in ("low", "medium", "xhigh", "high"): reasoning_effort = "xhigh" if effort_str == "high" else effort_str if enable_thinking is None: enable_thinking = True elif enable_thinking is not False: # Default to 'low' reasoning effort to prevent the model from overthinking if any(k in model_req for k in [":deep", ":high", ":xhigh"]): reasoning_effort = "xhigh" else: reasoning_effort = "low" chat_template_kwargs = {} if enable_thinking is not None: chat_template_kwargs["enable_thinking"] = bool(enable_thinking) if reasoning_effort is not None: chat_template_kwargs["reasoning_effort"] = str(reasoning_effort) res = {} if chat_template_kwargs: res["chat_template_kwargs"] = chat_template_kwargs if reasoning_budget is not None: res["reasoning_budget"] = reasoning_budget elif enable_thinking is False: res["reasoning_budget"] = 0 return res def _map_format(fmt, kw): if not fmt: return if fmt == "json": kw["response_format"] = {"type": "json_object"} elif isinstance(fmt, dict): kw["response_format"] = {"type": "json_object", "schema": fmt} def ollama_chat(payload): """POST /api/chat — translate ollama chat -> llama.cpp chat and back.""" stream = bool(payload.get("stream", False)) opts = payload.get("options", {}) or {} raw_msgs = payload.get("messages", []) top_images = payload.get("images", []) if top_images and raw_msgs: for m in reversed(raw_msgs): if m.get("role") == "user": m.setdefault("images", []).extend(top_images) break messages = [_openai_msg_from_ollama_message(m) for m in raw_msgs] # strip assistant messages that were purely tool responses placeholders messages = [m for m in messages if not (m.get("role") == "assistant" and not m.get("content"))] tools = _ollama_tools_to_openai(payload.get("tools")) kw = {} _map_options(opts, kw) _map_format(payload.get("format"), kw) think_params = _extract_thinking_params(payload) kw.update(think_params) if tools: kw["tools"] = tools if not stream: t0 = now_ns() data = _backend_chat(messages, **kw) t1 = now_ns() msg = data["choices"][0]["message"] usage = data.get("usage", {}) timings = data.get("timings", {}) done_reason = _map_done_reason(data["choices"][0].get("finish_reason")) om = { "role": "assistant", "content": msg.get("content") or "", } if EMIT_THINKING and msg.get("reasoning_content"): om["reasoning_content"] = msg["reasoning_content"] oc = _openai_tool_calls_to_ollama(msg.get("tool_calls")) if oc: om["tool_calls"] = oc p_count = usage.get("prompt_tokens", est_tokens(str(messages))) p_dur = int(timings.get("prompt_ms", 0) * 1_000_000) e_count = usage.get("completion_tokens", est_tokens(msg.get("content") or "")) e_dur = int(timings.get("predicted_ms", 0) * 1_000_000) tot_dur = t1 - t0 if e_dur == 0 and tot_dur > p_dur: e_dur = tot_dur - p_dur return { "model": MODEL_NAME, "created_at": now_iso(), "message": om, "done": True, "done_reason": done_reason, "total_duration": tot_dur, "load_duration": 0, "prompt_eval_count": p_count, "prompt_eval_duration": p_dur, "eval_count": e_count, "eval_duration": e_dur, }, False # streaming: emit NDJSON lines as backend chunks arrive def gen(): t0 = now_ns() usage = {} timings = {} eval_count = 0 finish_reason = "stop" tc_parts = {} # id -> {"name":..., "args": "..."} accumulated across chunks for chunk in _backend_stream(messages, **kw): if chunk == "[DONE]": break try: obj = json.loads(chunk) except json.JSONDecodeError: continue if obj.get("usage"): usage.update(obj["usage"]) if obj.get("timings"): timings.update(obj["timings"]) choices = obj.get("choices", []) if not choices: continue choice = choices[0] delta = choice.get("delta", {}) content = delta.get("content") reasoning = delta.get("reasoning_content") finish = choice.get("finish_reason") if finish: finish_reason = finish d_tc = delta.get("tool_calls") if d_tc: for tc in d_tc: tid = tc.get("id") or next(iter(tc_parts), None) or "call_" + uuid.uuid4().hex[:16] fn = tc.get("function", {}) entry = tc_parts.setdefault(tid, {"name": "", "args": ""}) if fn.get("name"): entry["name"] = fn["name"] if fn.get("arguments"): entry["args"] += fn["arguments"] continue if content: eval_count += 1 yield json.dumps({ "model": MODEL_NAME, "created_at": now_iso(), "message": {"role": "assistant", "content": content}, "done": False, }) + "\n" elif reasoning and EMIT_THINKING: eval_count += 1 yield json.dumps({ "model": MODEL_NAME, "created_at": now_iso(), "message": {"role": "assistant", "reasoning_content": reasoning}, "done": False, }) + "\n" # Emit final summary chunk with exact stats for OpenWebUI tok/s t1 = now_ns() tot_dur = t1 - t0 p_count = usage.get("prompt_tokens", est_tokens(str(messages))) p_dur = int(timings.get("prompt_ms", 0) * 1_000_000) e_count = usage.get("completion_tokens", eval_count) e_dur = int(timings.get("predicted_ms", 0) * 1_000_000) if e_dur == 0 and tot_dur > p_dur: e_dur = tot_dur - p_dur om = {"role": "assistant", "content": ""} if tc_parts: om["tool_calls"] = [ { "function": { "name": e["name"], "arguments": json.loads(e["args"]) if e["args"] else {}, } } for e in tc_parts.values() ] yield json.dumps({ "model": MODEL_NAME, "created_at": now_iso(), "message": om, "done": True, "done_reason": _map_done_reason(finish_reason), "total_duration": tot_dur, "load_duration": 0, "prompt_eval_count": p_count, "prompt_eval_duration": p_dur, "eval_count": e_count, "eval_duration": e_dur, }) + "\n" return gen(), True def ollama_generate(payload): """POST /api/generate — single-prompt completion.""" stream = bool(payload.get("stream", False)) opts = payload.get("options", {}) or {} prompt = payload.get("prompt", "") images = payload.get("images", []) system = payload.get("system") model_name = payload.get("model") or MODEL_NAME messages = [] if system: messages.append({"role": "system", "content": system}) if images: content_list = [] if prompt: content_list.append({"type": "text", "text": prompt}) for img_b64 in images: if not str(img_b64).startswith("data:"): img_url = f"data:image/jpeg;base64,{img_b64}" else: img_url = str(img_b64) content_list.append({"type": "image_url", "image_url": {"url": img_url}}) messages.append({"role": "user", "content": content_list}) else: messages.append({"role": "user", "content": prompt}) kw = {} _map_options(opts, kw) _map_format(payload.get("format"), kw) think_params = _extract_thinking_params(payload) kw.update(think_params) if not stream: t0 = now_ns() data = _backend_chat(messages, **kw) t1 = now_ns() msg = data["choices"][0]["message"] usage = data.get("usage", {}) timings = data.get("timings", {}) p_count = usage.get("prompt_tokens", est_tokens(prompt)) p_dur = int(timings.get("prompt_ms", 0) * 1_000_000) e_count = usage.get("completion_tokens", est_tokens(msg.get("content") or "")) e_dur = int(timings.get("predicted_ms", 0) * 1_000_000) tot_dur = t1 - t0 if e_dur == 0 and tot_dur > p_dur: e_dur = tot_dur - p_dur return { "model": MODEL_NAME, "created_at": now_iso(), "response": msg.get("content") or "", "done": True, "done_reason": _map_done_reason(data["choices"][0].get("finish_reason")), "context": [], "total_duration": tot_dur, "load_duration": 0, "prompt_eval_count": p_count, "prompt_eval_duration": p_dur, "eval_count": e_count, "eval_duration": e_dur, }, False def gen(): t0 = now_ns() usage = {} timings = {} eval_count = 0 finish_reason = "stop" for chunk in _backend_stream(messages, **kw): if chunk == "[DONE]": break try: obj = json.loads(chunk) except json.JSONDecodeError: continue if obj.get("usage"): usage.update(obj["usage"]) if obj.get("timings"): timings.update(obj["timings"]) choices = obj.get("choices", []) if not choices: continue choice = choices[0] delta = choice.get("delta", {}) finish = choice.get("finish_reason") if finish: finish_reason = finish content = delta.get("content") if content: eval_count += 1 yield json.dumps({ "model": MODEL_NAME, "created_at": now_iso(), "response": content, "done": False, }) + "\n" t1 = now_ns() tot_dur = t1 - t0 p_count = usage.get("prompt_tokens", est_tokens(prompt)) p_dur = int(timings.get("prompt_ms", 0) * 1_000_000) e_count = usage.get("completion_tokens", eval_count) e_dur = int(timings.get("predicted_ms", 0) * 1_000_000) if e_dur == 0 and tot_dur > p_dur: e_dur = tot_dur - p_dur yield json.dumps({ "model": MODEL_NAME, "created_at": now_iso(), "response": "", "done": True, "done_reason": _map_done_reason(finish_reason), "context": [], "total_duration": tot_dur, "load_duration": 0, "prompt_eval_count": p_count, "prompt_eval_duration": p_dur, "eval_count": e_count, "eval_duration": e_dur, }) + "\n" return gen(), True def ollama_embed(payload): """POST /api/embed — modern batch embeddings.""" t0 = now_ns() raw_input = payload.get("input") or payload.get("prompt") or "" model_name = payload.get("model") or MODEL_NAME inputs = [raw_input] if isinstance(raw_input, str) else list(raw_input) embeddings = [] p_count = 0 for text in inputs: try: with _post("/embeddings", {"model": MODEL_NAME, "input": text}) as r: data = json.loads(r.read().decode()) emb = data.get("data", [{}])[0].get("embedding", []) except Exception: emb = [0.0] * 5120 embeddings.append(emb) p_count += est_tokens(text) t1 = now_ns() return { "model": model_name, "embeddings": embeddings, "total_duration": t1 - t0, "load_duration": 0, "prompt_eval_count": p_count, }, False def ollama_embeddings_legacy(payload): """POST /api/embeddings — legacy single embedding.""" prompt = payload.get("prompt") or "" try: with _post("/embeddings", {"model": MODEL_NAME, "input": prompt}) as r: data = json.loads(r.read().decode()) emb = data.get("data", [{}])[0].get("embedding", []) except Exception: emb = [0.0] * 5120 return {"embedding": emb}, False def ollama_pull(payload): """POST /api/pull — stream download progress events for CLI/clients.""" def gen(): yield json.dumps({"status": "pulling manifest"}) + "\n" yield json.dumps({"status": "verifying sha256 digest"}) + "\n" yield json.dumps({"status": "writing manifest"}) + "\n" yield json.dumps({"status": "success"}) + "\n" return gen(), True # ---------------------------------------------------------------------------- # Anthropic Messages API # ---------------------------------------------------------------------------- def _anthropic_content_to_openai(content, is_last=False): """Convert Anthropic message content (string or block array) to OpenAI string.""" if isinstance(content, str): return content parts = [] for block in content: btype = block.get("type") if btype == "text": parts.append(block.get("text", "")) elif btype == "thinking": # keep reasoning in context if EMIT_THINKING: parts.append(block.get("thinking", "")) elif btype == "tool_result": tc = block.get("content", "") if isinstance(tc, list): tc = "".join(b.get("text", "") for b in tc if b.get("type") == "text") parts.append(f"[tool_result: {tc}]") elif btype == "image": parts.append("[image]") return "\n".join(parts) def _anthropic_tools_to_openai(tools): """Anthropic tools ({name,input_schema}) -> OpenAI tools.""" if not tools: return None out = [] for t in tools: out.append({ "type": "function", "function": { "name": t.get("name", ""), "description": t.get("description", ""), "parameters": t.get("input_schema", {}), }, }) return out def _openai_tool_calls_to_anthropic(tool_calls): """OpenAI tool_calls -> Anthropic tool_use content blocks.""" if not tool_calls: return [] blocks = [] for tc in tool_calls: fn = tc.get("function", {}) try: args = json.loads(fn.get("arguments", "{}")) except json.JSONDecodeError: args = {} blocks.append({ "type": "tool_use", "id": tc.get("id") or "toolu_" + uuid.uuid4().hex[:24], "name": fn.get("name", ""), "input": args, }) return blocks def _anthropic_messages_to_openai(msgs): out = [] for m in msgs: role = m.get("role") content = m.get("content", "") if role == "assistant": blocks = content if isinstance(content, list) else [] has_tool = any(b.get("type") == "tool_use" for b in blocks) text = _anthropic_content_to_openai(content) if has_tool and not text: text = "[tool_call]" out.append({"role": "assistant", "content": text}) else: out.append({"role": role, "content": _anthropic_content_to_openai(content)}) return out def anthropic_messages(payload): """POST /v1/messages — Anthropic Messages API shape.""" stream = bool(payload.get("stream", False)) system = payload.get("system") msgs = payload.get("messages", []) max_tokens = payload.get("max_tokens", 1024) temperature = payload.get("temperature") top_p = payload.get("top_p") openai_msgs = _anthropic_messages_to_openai(msgs) if system: sys_text = system if isinstance(system, str) else " ".join( b.get("text", "") for b in system if b.get("type") == "text" ) openai_msgs.insert(0, {"role": "system", "content": sys_text}) kw = {"max_tokens": max_tokens} if temperature is not None: kw["temperature"] = temperature if top_p is not None: kw["top_p"] = top_p tools = _anthropic_tools_to_openai(payload.get("tools")) if tools: kw["tools"] = tools think_params = _extract_thinking_params(payload) kw.update(think_params) msg_id = "msg_" + uuid.uuid4().hex[:24] if not stream: t0 = now_ns() data = _backend_chat(openai_msgs, **kw) t1 = now_ns() choice = data["choices"][0] msg = choice["message"] usage = data.get("usage", {}) content = [] if EMIT_THINKING and msg.get("reasoning_content"): content.append({"type": "thinking", "thinking": msg["reasoning_content"]}) tool_blocks = _openai_tool_calls_to_anthropic(msg.get("tool_calls")) if tool_blocks: content.extend(tool_blocks) if msg.get("content"): content.append({"type": "text", "text": msg.get("content")}) return { "id": msg_id, "type": "message", "role": "assistant", "model": MODEL_NAME, "content": content, "stop_reason": _anthropic_stop_reason(choice.get("finish_reason")), "stop_sequence": None, "usage": { "input_tokens": usage.get("prompt_tokens", est_tokens(str(openai_msgs))), "output_tokens": usage.get("completion_tokens", 0), }, }, False def gen(): # Emit Anthropic SSE event stream from llama.cpp chunks. def sse(event, data): return f"event: {event}\ndata: {json.dumps(data)}\n\n" yield sse("message_start", { "type": "message_start", "message": { "id": msg_id, "type": "message", "role": "assistant", "model": MODEL_NAME, "content": [], "stop_reason": None, "usage": {"input_tokens": est_tokens(str(openai_msgs)), "output_tokens": 0}, }, }) yield sse("content_block_start", { "type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}, }) text_acc = "" tc_idx = 0 tc_parts = {} # tool call id -> accumulated info for chunk in _backend_stream(openai_msgs, **kw): if chunk == "[DONE]": break try: obj = json.loads(chunk) except json.JSONDecodeError: continue choice = obj.get("choices", [{}])[0] delta = choice.get("delta", {}) content = delta.get("content") reasoning = delta.get("reasoning_content") finish = choice.get("finish_reason") d_tc = delta.get("tool_calls") if d_tc: # tool call fragments: emit tool_use block with input_json_delta for tc in d_tc: fn = tc.get("function", {}) if tc.get("id"): tc_idx += 1 current_tc_id = tc["id"] tc_parts[current_tc_id] = {"name": fn.get("name", ""), "args": "", "block_index": 1 + tc_idx - 1} yield sse("content_block_stop", {"type": "content_block_stop", "index": 0}) yield sse("content_block_start", { "type": "content_block_start", "index": tc_idx, "content_block": {"type": "tool_use", "id": "toolu_" + current_tc_id[:24], "name": fn.get("name", ""), "input": {}}, }) else: # argument fragment for the most recent tool call if tc_parts: tid = list(tc_parts)[-1] tc_parts[tid]["args"] += fn.get("arguments", "") continue if content: text_acc += content yield sse("content_block_delta", { "type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": content}, }) elif reasoning and EMIT_THINKING: yield sse("content_block_delta", { "type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": reasoning}, }) if finish: # close any open tool_use block(s) with accumulated input for tid, entry in tc_parts.items(): try: args = json.loads(entry["args"]) if entry["args"] else {} except json.JSONDecodeError: args = {} yield sse("content_block_delta", { "type": "content_block_delta", "index": entry["block_index"], "delta": {"type": "input_json_delta", "partial_json": json.dumps(args)}, }) yield sse("content_block_stop", { "type": "content_block_stop", "index": entry["block_index"], }) yield sse("content_block_stop", {"type": "content_block_stop", "index": 0}) yield sse("message_delta", { "type": "message_delta", "delta": {"stop_reason": _anthropic_stop_reason(finish), "stop_sequence": None}, "usage": {"output_tokens": est_tokens(text_acc)}, }) yield sse("message_stop", {"type": "message_stop"}) break return gen(), True def anthropic_count_tokens(payload): """POST /v1/messages/count_tokens — rough estimate.""" msgs = payload.get("messages", []) total = est_tokens(payload.get("system", "")) for m in msgs: total += est_tokens(str(m.get("content", ""))) return {"input_tokens": total}, False # ---------------------------------------------------------------------------- # Mapping helpers # ---------------------------------------------------------------------------- def _map_done_reason(fr): return { "stop": "stop", "length": "length", "tool_calls": "tool_calls", }.get(fr, "stop") def _anthropic_stop_reason(fr): return { "stop": "end_turn", "length": "max_tokens", "tool_calls": "tool_use", }.get(fr, "end_turn") # ---------------------------------------------------------------------------- # HTTP handler # ---------------------------------------------------------------------------- class Handler(http.server.BaseHTTPRequestHandler): server_version = f"OllamaProxy/{VERSION}" def log_message(self, fmt, *args): sys.stderr.write(f"[ollama-proxy] {self.command} {self.path} {fmt % args}\n") def _set_cors_headers(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS, HEAD") self.send_header("Access-Control-Allow-Headers", "*") self.send_header("Access-Control-Expose-Headers", "*") def _read_json(self): length = int(self.headers.get("Content-Length", 0)) if length == 0: return {} raw = self.rfile.read(length) try: return json.loads(raw.decode()) except (json.JSONDecodeError, UnicodeDecodeError): return {} def _send(self, code, obj_or_bytes, ctype="application/json"): if isinstance(obj_or_bytes, bytes): body = obj_or_bytes elif isinstance(obj_or_bytes, (dict, list)): body = json.dumps(obj_or_bytes).encode() else: body = str(obj_or_bytes).encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Ollama-Proxy", "llama.cpp") self._set_cors_headers() try: self.end_headers() self.wfile.write(body) except (BrokenPipeError, ConnectionResetError): pass def _send_stream(self, gen): self.send_response(200) self.send_header("Content-Type", "application/x-ndjson") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "close") self._set_cors_headers() self.end_headers() try: for line in gen: self.wfile.write(line.encode() if isinstance(line, str) else line) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass def _send_sse(self, gen): 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._set_cors_headers() self.end_headers() try: for chunk in gen: self.wfile.write(chunk.encode() if isinstance(chunk, str) else chunk) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass def do_OPTIONS(self): self.send_response(204) self._set_cors_headers() self.end_headers() def do_HEAD(self): path = self.path.split("?")[0] self.send_response(200) if path.startswith("/api/blobs/"): self.send_header("Content-Type", "application/octet-stream") else: self.send_header("Content-Type", "text/plain") self._set_cors_headers() self.send_header("Content-Length", "0") self.end_headers() # ---- routes ---- def do_GET(self): path = self.path.split("?")[0] if path == "/" or path == "": self._send(200, "Ollama is running", ctype="text/plain") elif path == "/api/version": self._send(200, ollama_version()) elif path == "/api/tags": self._send(200, ollama_tags()) elif path == "/api/ps": self._send(200, ollama_ps()) elif path == "/v1/models": self._send(200, { "object": "list", "data": [ { "id": tag, "object": "model", "owned_by": "local", "created": int(time.time()), "context_length": CTX_SIZE, "max_context_length": CTX_SIZE, "max_tokens": MAX_OUTPUT, "max_output_tokens": MAX_OUTPUT, } for tag in MODEL_TAGS ], }) elif path == "/health": self._send(200, {"status": "ok"}) elif path == "/api/status": self._send(200, ollama_status()) elif path == "/api/experimental/model-recommendations": self._send(200, ollama_recommendations()) else: self._send(404, {"error": f"unknown GET path {path}"}) def do_POST(self): path = self.path.split("?")[0] payload = self._read_json() try: if path == "/api/chat": result, is_stream = ollama_chat(payload) if is_stream: self._send_stream(result) else: self._send(200, result) elif path == "/api/generate": result, is_stream = ollama_generate(payload) if is_stream: self._send_stream(result) else: self._send(200, result) elif path == "/api/embed": result, _ = ollama_embed(payload) self._send(200, result) elif path == "/api/embeddings": result, _ = ollama_embeddings_legacy(payload) self._send(200, result) elif path == "/api/show": self._send(200, ollama_show(payload)) elif path == "/api/pull": if payload.get("stream", True): result, is_stream = ollama_pull(payload) self._send_stream(result) else: self._send(200, {"status": "success"}) elif path == "/api/push" or path == "/api/create" or path.startswith("/api/blobs/"): self._send(200, {"status": "success", "digest": "sha256:" + "0" * 64}) elif path == "/api/copy" or path == "/api/delete": self._send(200, {"status": "success"}) elif path == "/v1/messages": result, is_stream = anthropic_messages(payload) if is_stream: self._send_sse(result) else: self._send(200, result) elif path == "/v1/messages/count_tokens": result, _ = anthropic_count_tokens(payload) self._send(200, result) elif path in ("/v1/chat/completions", "/v1/completions", "/v1/embeddings", "/v1/responses"): body = json.dumps(payload).encode() backend_path = BACKEND + path[len("/v1"):] req = urllib.request.Request( backend_path, data=body, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req) as r: if payload.get("stream"): 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._set_cors_headers() self.end_headers() while True: line = r.readline() if not line: break self.wfile.write(line) self.wfile.flush() else: raw = r.read() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self._set_cors_headers() self.end_headers() self.wfile.write(raw) except urllib.error.HTTPError as e: self._send(e.code, {"error": e.read().decode(errors="replace")}) else: self._send(404, {"error": f"unknown POST path {path}"}) except urllib.error.HTTPError as e: try: detail = e.read().decode(errors="replace") except Exception: detail = "" self._send(e.code, {"error": detail or str(e)}) except Exception as e: self._send(500, {"error": f"{type(e).__name__}: {e}"}) do_PUT = do_POST do_DELETE = do_POST def main(): server = http.server.ThreadingHTTPServer((HOST, PORT), Handler) print(f"ollama-proxy listening on http://{HOST}:{PORT} -> {BACKEND}", flush=True) try: server.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()