1#!/usr/bin/env python3
2"""Adapt Responses tool namespaces to vLLM; 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 if (data.get("output_config") or {}).get("effort") in {"high", "max"}:
124 data["output_config"]["effort"] = "xhigh"
125 return data
126
127
128class Handler(BaseHTTPRequestHandler):
129 protocol_version = "HTTP/1.1"
130
131 def handle(self):
132 try:
133 super().handle()
134 except (ConnectionResetError, BrokenPipeError):
135 pass
136
137 def log_message(self, *args):
138 pass # Inference bodies and keys must not enter access logs.
139
140 def do_GET(self):
141 self.proxy()
142
143 def do_POST(self):
144 self.proxy()
145
146 def proxy(self):
147 connection = http.client.HTTPConnection("127.0.0.1", 8081, timeout=WAIT_SECONDS)
148 started = False
149 try:
150 length = int(self.headers.get("Content-Length", "0"))
151 if length < 0 or length > 20 * 1024 * 1024 or self.headers.get("Transfer-Encoding"):
152 self.send_error(413)
153 return
154 # Authenticate before allocating inference bodies.
155 if self.path != "/health":
156 key = os.environ["VLLM_API_KEY"]
157 supplied = self.headers.get("x-api-key", "") or self.headers.get("Authorization", "").removeprefix("Bearer ")
158 if not hmac.compare_digest(supplied, key):
159 self.send_error(401)
160 self.close_connection = True
161 return
162 body = self.rfile.read(length)
163 identities = {}
164 if self.command == "POST":
165 path = self.path.split("?", 1)[0]
166 if path == "/v1/responses":
167 data, identities = normalize(json.loads(body))
168 body = json.dumps(data).encode()
169 elif path in {"/v1/messages", "/v1/messages/count_tokens"}:
170 body = json.dumps(normalize_anthropic(json.loads(body))).encode()
171 headers = {key: val for key, val in self.headers.items() if key.lower() not in HOP_HEADERS | {"authorization", "x-api-key"}}
172 headers["Authorization"] = "Bearer " + os.environ["VLLM_API_KEY"]
173 connection.request(self.command, self.path, body=body or None, headers=headers)
174 response = connection.getresponse()
175 streaming = "text/event-stream" in response.getheader("Content-Type", "")
176 if not streaming:
177 content = response.read()
178 if identities and response.status == 200:
179 content = json.dumps(restore(json.loads(content), identities)).encode()
180 self.send_response(response.status)
181 for key, val in response.getheaders():
182 if key.lower() not in HOP_HEADERS:
183 self.send_header(key, val)
184 self.send_header("Content-Length", str(len(content)))
185 self.end_headers()
186 self.wfile.write(content)
187 return
188 self.send_response(response.status)
189 started = True
190 self.send_header("Content-Type", "text/event-stream")
191 self.send_header("Cache-Control", "no-cache")
192 self.send_header("Connection", "close")
193 self.end_headers()
194 self.close_connection = True
195 pending = b""
196 while chunk := response.read1(65536):
197 if not identities:
198 self.wfile.write(chunk)
199 self.wfile.flush()
200 continue
201 pending += chunk
202 while b"\n\n" in pending:
203 frame, pending = pending.split(b"\n\n", 1)
204 lines = []
205 for line in frame.split(b"\n"):
206 if line.startswith(b"data: ") and line[6:] != b"[DONE]":
207 line = b"data: " + json.dumps(restore(json.loads(line[6:]), identities)).encode()
208 lines.append(line)
209 self.wfile.write(b"\n".join(lines) + b"\n\n")
210 self.wfile.flush()
211 if pending:
212 self.wfile.write(pending)
213 except (BrokenPipeError, ConnectionResetError):
214 pass
215 except (ValueError, KeyError, TypeError) as error:
216 if not started:
217 self.send_error(400, str(error))
218 self.close_connection = True
219 except (OSError, http.client.HTTPException):
220 if not started:
221 self.send_error(502)
222 self.close_connection = True
223 finally:
224 connection.close()
225
226
227if __name__ == "__main__":
228 backend = subprocess.Popen(["bash", "/app/single-user/start_qwen.sh"])
229 server = ThreadingHTTPServer(("0.0.0.0", 8080), Handler)
230 def stop(signum, frame):
231 if backend.poll() is None:
232 backend.terminate()
233 signal.signal(signal.SIGTERM, stop)
234 signal.signal(signal.SIGINT, stop)
235 def monitor():
236 backend.wait()
237 server.shutdown()
238 threading.Thread(target=monitor, daemon=True).start()
239 try:
240 server.serve_forever()
241 finally:
242 server.server_close()
243 if backend.poll() is None:
244 backend.terminate()
245 sys.exit(backend.wait())