import requests,json
from flask import Flask,Response,request,stream_with_context,jsonify
from datetime import datetime
import time,os
app = Flask(__name__)
gh_token = input("https://github.com/settings/personal-access-tokens/new\n\ntoken:")
def get_headers(gh_token):
return {
"Content-Type":"application/json",
"Authorization":f"Bearer {gh_token}",
"key":"NfZFVegMpdyT3P5UmAggr7T7Hb6PlcbB",
"timestamp":str(int(time.time()*1000))[:-2]+str((datetime.now().minute//2)+10),
"v":"1",
"extensionVersion":"0.8.66",
"os": os.name,
"uniqueId": "NOT_UNIQUE"
}
def chat_request(messages, temp, max_tokens, top_p, stop):
response = requests.post(
"https://proxy-server-blue-l6vsfbzhba-uw.a.run.app/stream_chat",
headers=get_headers(gh_token),
json={
"messages": messages,
"model": "claude-3-5-sonnet-latest",
"temperature": temp,
"top_p": top_p,
"max_tokens": max_tokens,
"stop": stop
},
stream = True
)
if response.ok:
return response, None
else:
return None, response.text
@app.route("/messages", methods=["POST"])
def handle_chat():
data = request.json
streaming = data.get("stream", True)
messages = []
if data.get("system"):
messages.append({"role": "system", "content": data["system"][0]["text"]})
for msg in data.get("messages", []):
for content in msg["content"]:
messages.append({"role": msg.get("role"), "content": content.get("text")})
# no prefill
if messages and messages[-1]["role"] == "assistant":
messages.append({"role": "user", "content":"."})
result, error = chat_request(
messages=messages,
temp=data.get("temperature"),
max_tokens=data.get("max_tokens"),
top_p=data.get("top_p"),
stop=data.get("stop_sequences")
)
if error:
return jsonify({
"error": {
"message": error,
"code": "bad_request"
}
}), 400
if streaming:
def generate():
first_line = True
for l in result.iter_lines():
if not l:
continue
try:
line = l.decode('utf-8').strip()
if not line.startswith('{'):
resp={
'type':'content_block_delta',
'delta':{'type': 'text_delta', 'text':'\n\n'+line if not first_line else line},
'index':0
}
yield f"event: content_block_delta\ndata: {json.dumps(resp)}\n\n"
first_line = False
continue
d = json.loads(line)
if'content'in d:
resp={
'type':'content_block_delta',
'delta':{'type': 'text_delta', 'text': d['content']},
'index':0
}
yield f"event: content_block_delta\ndata: {json.dumps(resp)}\n\n"
if d.get('choices',[{}])[0].get('finish_reason') is not None:
yield 'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null}}\n\n'
yield 'event: message_stop\ndata: {"type":"message_stop"}\n\n'
break
except:continue
return Response(stream_with_context(generate()), content_type='text/event-stream', headers={'Cache-Control':'no-cache','Connection':'keep-alive'})
else:
return {"type": "message", "content": [{"type": "text", "text": result.text}]}
app.run()