| """No-framework local HTTP runtime. Intended for localhost or a separately secured host, not Static Spaces.""" |
| import argparse,json |
| from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer |
| from .runtime import load_model,respond,proposed_plan |
| class App: |
| ares=None;xiphos=None |
| class Handler(BaseHTTPRequestHandler): |
| def reply(self,status,obj): |
| raw=json.dumps(obj).encode();self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Access-Control-Allow-Origin','*');self.send_header('Content-Length',str(len(raw)));self.end_headers();self.wfile.write(raw) |
| def do_OPTIONS(self):self.send_response(204);self.send_header('Access-Control-Allow-Origin','*');self.send_header('Access-Control-Allow-Headers','Content-Type');self.end_headers() |
| def do_GET(self): |
| if self.path=='/health':return self.reply(200,{'ok':True,'ares_loaded':bool(App.ares),'xiphos_loaded':bool(App.xiphos),'execution':'disabled'}) |
| self.reply(404,{'error':'not found'}) |
| def do_POST(self): |
| if self.path not in ('/chat','/xiphos/plan'):return self.reply(404,{'error':'not found'}) |
| try: |
| n=int(self.headers.get('Content-Length','0')); body=json.loads(self.rfile.read(min(n,100_000)));message=str(body['message']).strip() |
| if not message or len(message)>20_000:raise ValueError('message must contain 1–20,000 characters') |
| if self.path=='/chat':return self.reply(200,{'role':'Ares','response':respond('Ares',App.ares,message)}) |
| return self.reply(200,proposed_plan(App.xiphos,message)) |
| except (ValueError,KeyError,json.JSONDecodeError) as e:self.reply(400,{'error':str(e)}) |
| def log_message(self,*args):pass |
| def main(): |
| p=argparse.ArgumentParser();p.add_argument('--ares',required=True);p.add_argument('--xiphos',required=True);p.add_argument('--tokenizer',required=True);p.add_argument('--host',default='127.0.0.1');p.add_argument('--port',type=int,default=8787);a=p.parse_args() |
| App.ares=load_model(a.ares,a.tokenizer,expected_role='ares');App.xiphos=load_model(a.xiphos,a.tokenizer,expected_role='xiphos') |
| print(f'Local runtime at http://{a.host}:{a.port}; tool execution is disabled.') |
| ThreadingHTTPServer((a.host,a.port),Handler).serve_forever() |
| if __name__=='__main__':main() |
|
|
