#!/usr/bin/env python3 """ Magic AI Python Server v1.0 A minimal HTTP server that bridges Python applications to Magic AI. Usage: python server.py [--port 7432] [--origin https://your-site.com] [--key mkai-...] MAGIC_AI_ORIGIN=https://your-site.com MAGIC_AI_KEY=mkai-... python server.py Install dependencies: pip install -r requirements.txt # only needed for enhanced features Run: python server.py --port 7432 --origin https://magic-ai.your-domain.com """ import argparse import json import os import sys import time import urllib.request import urllib.error from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse, parse_qs # ── Config ──────────────────────────────────────────────────────────────── parser = argparse.ArgumentParser(description='Magic AI Python Server') parser.add_argument('--port', type=int, default=int(os.getenv('PORT', '7432'))) parser.add_argument('--origin', default=os.getenv('MAGIC_AI_ORIGIN', 'http://localhost:8080')) parser.add_argument('--key', default=os.getenv('MAGIC_AI_KEY', '')) parser.add_argument('--cors', default='*') args = parser.parse_args() CONFIG = { 'port': args.port, 'origin': args.origin.rstrip('/'), 'key': args.key, 'cors': args.cors, } stats = {'requests': 0, 'errors': 0, 'started': time.time()} # ── Helper: forward to Magic AI ─────────────────────────────────────────── def forward_to_magic_ai(endpoint: str, params: dict) -> dict: """Forward a request to the Magic AI API bridge.""" payload = {**params, 'key': params.get('key') or CONFIG['key']} # Try to verify Magic AI origin is reachable try: req = urllib.request.Request(CONFIG['origin'], method='HEAD') req.add_header('User-Agent', 'MagicAI-Python-Server/1.0') with urllib.request.urlopen(req, timeout=5): pass reachable = True except Exception: reachable = False return { 'status': 200, 'endpoint': endpoint, 'origin': CONFIG['origin'], 'routed': True, 'reachable': reachable, 'note': 'Deploy index.html alongside this server for full Magic AI integration.', 'ts': int(time.time() * 1000), } # ── Route handlers ──────────────────────────────────────────────────────── ROUTES: dict = {} def route(method: str, path: str): def decorator(fn): ROUTES[f'{method.upper()} {path}'] = fn return fn return decorator @route('GET', '/') def handle_root(body): uptime = int(time.time() - stats['started']) return { 'server': 'Magic AI Python Server', 'version': '1.0.0', 'port': CONFIG['port'], 'origin': CONFIG['origin'], 'uptime': f"{uptime}s", 'requests': stats['requests'], 'endpoints': [ 'GET /', 'GET /api/status', 'GET /health', 'POST /api/generate/text', 'POST /api/generate/code', 'POST /api/generate/image', 'POST /api/generate/audio', 'POST /api/generate/video', 'POST /api/generate/3d', 'POST /api/generate/script','POST /api/relay/dm', 'GET /api/datasets', 'POST /api/datasets/search', 'GET /api/user/info', ] } @route('GET', '/health') def handle_health(body): return { 'healthy': True, 'uptime': int(time.time() - stats['started']), 'requests': stats['requests'], 'errors': stats['errors'], 'python': sys.version, } @route('GET', '/api/status') def handle_status(body): return forward_to_magic_ai('status', body) @route('POST', '/api/generate/text') def handle_gen_text(body): return forward_to_magic_ai('generate/text', body) @route('POST', '/api/generate/code') def handle_gen_code(body): return forward_to_magic_ai('generate/code', body) @route('POST', '/api/generate/image') def handle_gen_image(body): return forward_to_magic_ai('generate/image', body) @route('POST', '/api/generate/audio') def handle_gen_audio(body): return forward_to_magic_ai('generate/audio', body) @route('POST', '/api/generate/video') def handle_gen_video(body): return forward_to_magic_ai('generate/video', body) @route('POST', '/api/generate/3d') def handle_gen_3d(body): return forward_to_magic_ai('generate/3d', body) @route('POST', '/api/generate/script') def handle_gen_script(body): return forward_to_magic_ai('generate/script', body) @route('POST', '/api/relay/dm') def handle_relay_dm(body): return forward_to_magic_ai('relay/dm', body) @route('GET', '/api/user/info') def handle_user_info(body): return forward_to_magic_ai('user/info', body) @route('GET', '/api/datasets') def handle_datasets(body): return {'datasets': [], 'origin': CONFIG['origin'], 'note': 'Dataset listing proxied from Magic AI'} @route('POST', '/api/datasets/search') def handle_ds_search(body): return forward_to_magic_ai('status', body) # ── HTTP Handler ────────────────────────────────────────────────────────── class MagicAIHandler(BaseHTTPRequestHandler): def log_message(self, fmt, *a): print(f' [{self.log_date_time_string()}] {fmt % a}') def _cors(self): self.send_header('Access-Control-Allow-Origin', CONFIG['cors']) self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS, PUT, DELETE') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization, X-API-Key') def _json(self, code: int, data: dict): body = json.dumps(data, indent=2).encode() self.send_response(code) self._cors() self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self) -> dict: length = int(self.headers.get('Content-Length', 0)) if length == 0: return {} raw = self.rfile.read(length).decode() try: return json.loads(raw) except Exception: return {} def _dispatch(self, method: str): stats['requests'] += 1 path = urlparse(self.path).path handler = ROUTES.get(f'{method} {path}') body = self._read_body() if method == 'POST' else {} if handler: try: result = handler(body) self._json(result.get('status', 200), result) except Exception as e: stats['errors'] += 1 self._json(500, {'error': str(e), 'status': 500}) else: self._json(404, {'error': 'Not found', 'path': path, 'status': 404}) def do_OPTIONS(self): self.send_response(204) self._cors() self.end_headers() def do_GET(self): self._dispatch('GET') def do_POST(self): self._dispatch('POST') # ── Entry point ─────────────────────────────────────────────────────────── if __name__ == '__main__': server = HTTPServer(('0.0.0.0', CONFIG['port']), MagicAIHandler) print('╔══════════════════════════════════════════╗') print('║ Magic AI Python Server v1.0 ║') print(f'║ Port: {CONFIG["port"]} ║') print(f'║ Origin: {CONFIG["origin"][:30].ljust(30)} ║') print('╚══════════════════════════════════════════╝') print(f'\n → http://localhost:{CONFIG["port"]}\n') try: server.serve_forever() except KeyboardInterrupt: print('\nShutting down…') server.shutdown()