#!/usr/bin/env python3 """mb.py — BLOBBO client for the MODELBEAST farm (m3ultra central queue :8777). Contract per ~/Documents/MESHGOD/scripts/mb_recon.py (verified 2026-07-16): Bearer MB_TOKEN (env, else ~/Documents/backnforth/.env), POST /api/jobs, poll GET /api/jobs/{id}, retrieve own assets by parent_job (never newest-glob — concurrent jobs race). Job JSON may carry raw control chars — strip pre-parse. Guest tier: max 4 active jobs, enforced here with a submit window. Usage: mb.py gen "" out.png one flux_local image mb.py batch prompts.json outdir/ [{"name":..,"prompt":..,"operator"?,"params"?}] mb.py img2glb in.png out.glb [operator] image -> mesh (default hunyuan3d_mlx) mb.py jobs list my recent jobs """ import json, mimetypes, os, sys, time, urllib.request, urllib.error, uuid HOST = os.environ.get('MB_HOST', 'http://100.89.131.57:8777') MAX_ACTIVE = 4 def token(): t = os.environ.get('MB_TOKEN') if t: return t for line in open(os.path.expanduser('~/Documents/backnforth/.env')): if line.startswith('MB_TOKEN='): return line.split('=', 1)[1].strip() sys.exit('no MB_TOKEN') def req(path, data=None, headers=None, raw=False): h = {'Authorization': f'Bearer {token()}'} h.update(headers or {}) r = urllib.request.Request(HOST + path, data=data, headers=h) body = urllib.request.urlopen(r, timeout=180).read() if raw: return body s = body.decode('utf-8', 'replace') return json.loads(''.join(c if c >= ' ' or c in '\t' else ' ' for c in s)) def submit(operator, params=None, asset_id=None): """Returns job id, or None on 429 (farm throttles submits harder than the 4-active contract — observed 429 on the 4th rapid submit, 2026-07-17).""" payload = {'operator': operator, 'params': params or {}} if asset_id: payload['asset_id'] = asset_id try: j = req('/api/jobs', data=json.dumps(payload).encode(), headers={'Content-Type': 'application/json'}) except urllib.error.HTTPError as e: if e.code == 429: return None raise return j['id'] def status(jid): return req(f'/api/jobs/{jid}').get('status') def upload(path): boundary = uuid.uuid4().hex ctype = mimetypes.guess_type(path)[0] or 'application/octet-stream' name = os.path.basename(path) body = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n' f'Content-Type: {ctype}\r\n\r\n').encode() + open(path, 'rb').read() + f'\r\n--{boundary}--\r\n'.encode() a = req('/api/assets', data=body, headers={'Content-Type': f'multipart/form-data; boundary={boundary}'}) return a.get('id') or (a.get('items') or [a])[0].get('id') def outputs_for(jid, suffix=None): a = req('/api/assets?limit=200') items = a if isinstance(a, list) else a.get('items', []) mine = [x for x in items if x.get('parent_job') == jid] if suffix: mine = [x for x in mine if str(x.get('filename', x.get('name', ''))).endswith(suffix)] return mine def fetch(asset_id, out): data = req(f'/api/assets/{asset_id}/file', raw=True) open(out, 'wb').write(data) return len(data) def wait(jid, label='', interval=6): t0 = time.time() while True: st = status(jid) if st in ('done', 'error', 'cancelled'): print(f'[mb] {label or jid}: {st} ({int(time.time()-t0)}s)', flush=True) return st time.sleep(interval) def cmd_gen(prompt, out): jid = submit('flux_local', {'prompt': prompt}) if wait(jid, out) != 'done': sys.exit(1) outs = outputs_for(jid) if not outs: sys.exit('[mb] no output asset for job') n = fetch(outs[0]['id'], out) print(f'[mb] wrote {out} ({n//1024}KB)') def cmd_batch(spec_path, outdir): jobs = json.load(open(spec_path)) os.makedirs(outdir, exist_ok=True) done_names = {os.path.splitext(f)[0] for f in os.listdir(outdir)} jobs = [j for j in jobs if j['name'] not in done_names] if done_names: print(f'[mb] resume: skipping {len(done_names)} already in {outdir}', flush=True) pending, active, results = list(jobs), {}, {} while pending or active: while pending and len(active) < MAX_ACTIVE: job = pending[0] params = dict(job.get('params', {})) if 'prompt' in job: params.setdefault('prompt', job['prompt']) jid = submit(job.get('operator', 'flux_local'), params) if jid is None: print('[mb] throttled (429), backing off 15s', flush=True) time.sleep(15) break pending.pop(0) active[jid] = job print(f'[mb] submitted {job["name"]} -> {jid}', flush=True) time.sleep(6) for jid in list(active): st = status(jid) if st in ('done', 'error', 'cancelled'): job = active.pop(jid) results[job['name']] = st if st == 'done': outs = outputs_for(jid) if outs: name = str(outs[0].get('filename', outs[0].get('name', 'out'))) ext = os.path.splitext(name)[1] or '.png' out = os.path.join(outdir, job['name'] + ext) n = fetch(outs[0]['id'], out) print(f'[mb] {job["name"]}: done -> {out} ({n//1024}KB)', flush=True) else: print(f'[mb] {job["name"]}: done but no asset visible', flush=True) results[job['name']] = 'done-no-asset' else: print(f'[mb] {job["name"]}: {st}', flush=True) bad = {k: v for k, v in results.items() if v != 'done'} print(f'[mb] batch complete: {len(results)-len(bad)}/{len(results)} ok' + (f' — failed: {bad}' if bad else '')) def cmd_img2glb(img, out, operator='hunyuan3d_mlx'): aid = upload(img) print(f'[mb] uploaded {os.path.basename(img)} -> asset {aid}') jid = submit(operator, {}, asset_id=aid) if wait(jid, out, interval=10) != 'done': sys.exit(1) outs = outputs_for(jid, '.glb') if not outs: sys.exit('[mb] no glb for job') n = fetch(outs[0]['id'], out) print(f'[mb] wrote {out} ({n//1024}KB)') def cmd_jobs(): j = req('/api/jobs?limit=20') items = j if isinstance(j, list) else j.get('items', []) for x in items: print(x.get('id'), x.get('operator'), x.get('status')) if __name__ == '__main__': cmd = sys.argv[1] if len(sys.argv) > 1 else '' if cmd == 'gen': cmd_gen(sys.argv[2], sys.argv[3]) elif cmd == 'batch': cmd_batch(sys.argv[2], sys.argv[3]) elif cmd == 'img2glb': cmd_img2glb(*sys.argv[2:5]) elif cmd == 'jobs': cmd_jobs() else: sys.exit(__doc__)