1#!/usr/bin/env python3
2"""Adapt current Responses tool namespaces to llama.cpp; forward other APIs unchanged."""
3import hmac
4import http.client
5from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
6import json
7import os
8import signal
9import subprocess
10import sys
11import threading
12
13WAIT_SECONDS = 8 * 60 * 60
14HOP_HEADERS = {"connection", "transfer-encoding", "content-length", "host", "accept-encoding"}
15
16
17def normalize(request):
18 data = request
19 identities = {}
20 tools = []
21
22 def add(tool, namespace=None):
23 kind = tool.get("type")
24 if kind not in {"function", "custom"}:
25 raise ValueError(f"Unsupported Responses tool type: {kind}")
26 original = tool["name"]
27 name = f"{namespace}__{original}" if namespace else original
28 if name in identities:
29 raise ValueError(f"Duplicate tool name: {name}")
30 identities[name] = (namespace, original, kind)
31 if kind == "custom":
32 description = tool.get("description", "") + " Pass the raw tool input as the input string."
33 grammar = tool.get("format", {})
34 if grammar.get("type") == "grammar":
35 description += " The input must follow this grammar:\n" + grammar["definition"]
36 tool = {
37 "type": "function", "name": name,
38 "description": description,
39 "parameters": {"type": "object", "properties": {"input": {"type": "string"}}, "required": ["input"]},
40 }
41 else:
42 tool["name"] = name
43 tools.append(tool)
44
45 for tool in data.get("tools", []):
46 if tool.get("type") == "namespace":
47 for nested in tool.get("tools", []):
48 add(nested, tool["name"])
49 else:
50 add(tool)
51 data["tools"] = tools
52 incoming = data.get("input", [])
53 if isinstance(incoming, str):
54 incoming = [{"role": "user", "content": incoming}]
55 instructions = []
56 if data.get("instructions"):
57 instructions.append(data.pop("instructions"))
58 messages = []
59 for item in incoming:
60 if item.get("role") in {"system", "developer"}:
61 content = item.get("content", "")
62 instructions.append(content if isinstance(content, str) else "\n".join(part.get("text", "") for part in content))
63 continue
64 if item.get("type") in {"function_call", "custom_tool_call"}:
65 namespace = item.pop("namespace", None)
66 if namespace:
67 item["name"] = namespace + "__" + item["name"]
68 if item["type"] == "custom_tool_call":
69 item["type"] = "function_call"
70 item["arguments"] = json.dumps({"input": item.pop("input")})
71 elif item.get("type") == "custom_tool_call_output":
72 item["type"] = "function_call_output"
73 messages.append(item)
74 if instructions:
75 messages.insert(0, {"role": "system", "content": "\n\n".join(instructions)})
76 data["input"] = messages
77 if isinstance(data.get("tool_choice"), dict):
78 choice = data["tool_choice"]
79 namespace = choice.pop("namespace", None)
80 if namespace:
81 choice["name"] = namespace + "__" + choice["name"]
82 if choice.get("type") == "custom":
83 choice["type"] = "function"
84 return data, identities
85
86
87def restore(value, identities):
88 if isinstance(value, list):
89 return [restore(item, identities) for item in value]
90 if not isinstance(value, dict):
91 return value
92 value = {key: restore(item, identities) for key, item in value.items()}
93 if value.get("type") == "function_call" and value.get("name") in identities:
94 namespace, name, kind = identities[value["name"]]
95 value["name"] = name
96 if namespace:
97 value["namespace"] = namespace
98 if kind == "custom":
99 value["type"] = "custom_tool_call"
100 arguments = value.pop("arguments", "")
101 value["input"] = json.loads(arguments).get("input", "") if arguments else ""
102 return value
103
104
105def normalize_anthropic(request):
106 data = request
107 def text(content):
108 if isinstance(content, str):
109 return content
110 if any(part.get("type") != "text" for part in content):
111 raise ValueError("Anthropic system content must contain text blocks.")
112 return "\n\n".join(part["text"] for part in content)
113 instructions = [text(data["system"])] if data.get("system") else []
114 messages = []
115 for message in data.get("messages", []):
116 if message.get("role") in {"system", "developer"}:
117 instructions.append(text(message["content"]))
118 else:
119 messages.append(message)
120 if instructions:
121 data["system"] = "\n\n".join(instructions)
122 data["messages"] = messages
123 return data
124
125
126class Handler(BaseHTTPRequestHandler):
127 protocol_version = "HTTP/1.1"
128
129 def handle(self):
130 try:
131 super().handle()
132 except (ConnectionResetError, BrokenPipeError):
133 pass
134
135 def log_message(self, *args):
136 pass # Inference bodies and keys must not enter access logs.
137
138 def do_GET(self):
139 self.proxy()
140
141 def do_POST(self):
142 self.proxy()
143
144 def proxy(self):
145 connection = http.client.HTTPConnection("127.0.0.1", 8081, timeout=WAIT_SECONDS)
146 started = False
147 try:
148 length = int(self.headers.get("Content-Length", "0"))
149 if length < 0 or length > 20 * 1024 * 1024 or self.headers.get("Transfer-Encoding"):
150 self.send_error(413)
151 return
152 # Authenticate before allocating inference bodies.
153 if self.path != "/health":
154 key = os.environ["LLAMA_API_KEY"]
155 supplied = self.headers.get("x-api-key", "") or self.headers.get("Authorization", "").removeprefix("Bearer ")
156 if not hmac.compare_digest(supplied, key):
157 self.send_error(401)
158 self.close_connection = True
159 return
160 body = self.rfile.read(length)
161 identities = {}
162 if self.command == "POST":
163 path = self.path.split("?", 1)[0]
164 if path == "/v1/responses":
165 data, identities = normalize(json.loads(body))
166 body = json.dumps(data).encode()
167 elif path in {"/v1/messages", "/v1/messages/count_tokens"}:
168 body = json.dumps(normalize_anthropic(json.loads(body))).encode()
169 headers = {key: val for key, val in self.headers.items() if key.lower() not in HOP_HEADERS}
170 connection.request(self.command, self.path, body=body or None, headers=headers)
171 response = connection.getresponse()
172 streaming = "text/event-stream" in response.getheader("Content-Type", "")
173 if not streaming:
174 content = response.read()
175 if identities and response.status == 200:
176 content = json.dumps(restore(json.loads(content), identities)).encode()
177 self.send_response(response.status)
178 for key, val in response.getheaders():
179 if key.lower() not in HOP_HEADERS:
180 self.send_header(key, val)
181 self.send_header("Content-Length", str(len(content)))
182 self.end_headers()
183 self.wfile.write(content)
184 return
185 self.send_response(response.status)
186 started = True
187 self.send_header("Content-Type", "text/event-stream")
188 self.send_header("Cache-Control", "no-cache")
189 self.send_header("Connection", "close")
190 self.end_headers()
191 self.close_connection = True
192 pending = b""
193 while chunk := response.read1(65536):
194 if not identities:
195 self.wfile.write(chunk)
196 self.wfile.flush()
197 continue
198 pending += chunk
199 while b"\n\n" in pending:
200 frame, pending = pending.split(b"\n\n", 1)
201 lines = []
202 for line in frame.split(b"\n"):
203 if line.startswith(b"data: ") and line[6:] != b"[DONE]":
204 line = b"data: " + json.dumps(restore(json.loads(line[6:]), identities)).encode()
205 lines.append(line)
206 self.wfile.write(b"\n".join(lines) + b"\n\n")
207 self.wfile.flush()
208 if pending:
209 self.wfile.write(pending)
210 except (BrokenPipeError, ConnectionResetError):
211 pass
212 except (ValueError, KeyError, TypeError) as error:
213 if not started:
214 self.send_error(400, str(error))
215 self.close_connection = True
216 except (OSError, http.client.HTTPException):
217 if not started:
218 self.send_error(502)
219 self.close_connection = True
220 finally:
221 connection.close()
222
223
224if __name__ == "__main__":
225 backend = subprocess.Popen(["/app/llama-server", *sys.argv[1:]])
226 server = ThreadingHTTPServer(("0.0.0.0", 8080), Handler)
227 def stop(signum, frame):
228 if backend.poll() is None:
229 backend.terminate()
230 signal.signal(signal.SIGTERM, stop)
231 signal.signal(signal.SIGINT, stop)
232 def monitor():
233 backend.wait()
234 server.shutdown()
235 threading.Thread(target=monitor, daemon=True).start()
236 try:
237 server.serve_forever()
238 finally:
239 server.server_close()
240 if backend.poll() is None:
241 backend.terminate()
242 sys.exit(backend.wait())