# AI Prompt Hub Kit Build one front door for the AI on your own server: a web page, a terminal command and desktop shortcuts that send each prompt to Claude Code first, ChatGPT (through the Codex command line) when Claude is out of quota, and local models for triage. Every prompt becomes a plan first; nothing runs until you say so. Your phone buzzes when a job ends. This is the generalized version of the hub described in [Build Your Own AI Prompt Hub](https://jwatte.com/blog/build-your-own-ai-prompt-hub/). Names, hostnames and paths are placeholders. Phone alerts are covered in the companion [ntfy Phone Alerts Cookbook](https://jwatte.com/downloads/ntfy-phone-alerts-cookbook.md). ## 1. What you need - A Linux server you control (any size; local models are optional). - Claude Code and the Codex CLI, installed and signed in **as the account the hub will run as**. Subscriptions (Claude Pro or Max, ChatGPT plans) work through these command lines' own sign-in. Claude Code's `--bare` mode does not use the subscription login, so don't use it here. - Optional: Ollama for local models, LiteLLM if you want one OpenAI-style endpoint for several, and a vector store such as Qdrant for notes. - Python 3.11 or later with `fastapi`, `uvicorn` and `pyjwt[crypto]`. - A Cloudflare account with a Tunnel and Access (or any reverse proxy that puts a sign-in in front and lets you verify who signed in). - An ntfy topic for alerts. ## 2. How it fits together ``` phone / browser ──► Cloudflare Access ──► tunnel ──┐ terminal (local token) ───────────────────────────┤ desktop icon (forced-command SSH key) ────────────┤ ▼ hub on 127.0.0.1:8790 plan: triage (small local model) + notes lookup run: Claude Code ─► Codex ─► local model (fallback order) in a per-job folder, on this server or another machine done: log + result + ntfy alert (job id and link only) ``` ## 3. The server, piece by piece The snippets below are the load-bearing parts. A complete server is about 600 lines including status checks and Remote Control; start with these and grow it. ### 3.1 Configuration ```python CFG = { "team_domain": "https://YOUR-TEAM.cloudflareaccess.com", "access_aud": None, # pinned on the first verified request (or set it from the Access app) "allowed_emails": ["you@example.com"], "engine_order": ["claude", "codex", "local"], "max_parallel": 2, "ask_tools": ["Read", "Grep", "Glob", "WebSearch", "WebFetch"], "ntfy_url": None, # read from an environment variable or a 0600 file, never from code } ``` ### 3.2 Who is asking: verify the Access token yourself The tunnel connector runs on the same machine, so every forwarded request arrives from 127.0.0.1. Loopback proves nothing. Verify the signed token Cloudflare Access attaches to every request: ```python import hmac, jwt from fastapi import HTTPException, Request JWKS = jwt.PyJWKClient(CFG["team_domain"] + "/cdn-cgi/access/certs", cache_keys=True) LOCAL_TOKEN = open("/home/you/.config/hub/token").read().strip() # mode 0600, for the terminal client def who(request: Request) -> str: tok = request.headers.get("x-hub-token") if tok and hmac.compare_digest(tok, LOCAL_TOKEN): return "local" assertion = request.headers.get("cf-access-jwt-assertion") if not assertion: raise HTTPException(401, "sign in through the hub's hostname") key = JWKS.get_signing_key_from_jwt(assertion).key claims = jwt.decode(assertion, key, algorithms=["RS256"], issuer=CFG["team_domain"], audience=CFG["access_aud"], options={"verify_aud": bool(CFG["access_aud"])}) email = (claims.get("email") or "").lower() if email not in CFG["allowed_emails"]: raise HTTPException(403, "not an allowed account") return email ``` Set `access_aud` to the Access application's AUD tag (Access, your application, Additional settings), or pin it from the first verified token and save it. ### 3.3 Plan first, run second ```python import json, secrets, sqlite3 @app.post("/api/plan") async def plan(request: Request): user = who(request) if not request.headers.get("content-type", "").startswith("application/json"): raise HTTPException(415, "JSON only") body = await request.json() prompt = str(body.get("prompt", "")).strip()[:20000] p = make_plan(prompt, body.get("target", "server"), body.get("engine", "auto"), body.get("mode")) jid, run_token = new_job_id(), secrets.token_urlsafe(18) save_job(jid, user, prompt, p, run_token, status="planned") return {"id": jid, "plan": p, "run_token": run_token} @app.post("/api/jobs/{jid}/run") async def run(jid: str, request: Request): who(request) body = await request.json() job = load_job(jid) if job["status"] != "planned" or not hmac.compare_digest(body.get("run_token", ""), job["run_token"] or ""): raise HTTPException(403, "not runnable") set_status(jid, "queued", run_token=None) # the token works once return {"id": jid, "status": "queued"} ``` Rules that keep this safe: a GET never creates or runs work (the page reads `?q=` and calls `/api/plan` itself); every state-changing request is JSON; send no CORS headers; the run token comes only in the plan response. `make_plan` does three cheap things: ask a small local model to classify the request, search your notes for the best few matches (titles and file paths), and pick the first engine that isn't cooling down. ### 3.4 The engines Claude Code, streaming JSON, prompt on stdin. In ask mode it gets read-only tools; in work mode it uses its auto permission mode inside the job folder. Never pass `--dangerously-skip-permissions`. ```python async def run_claude(jid, job, preamble, workdir): args = ["claude", "-p", "--output-format", "stream-json", "--verbose", "--append-system-prompt", preamble] if job["mode"] == "ask": args += ["--permission-mode", "default", "--allowedTools", " ".join(CFG["ask_tools"])] else: args += ["--permission-mode", "auto"] rc, lines = await stream(jid, args, stdin_text=job["prompt"], cwd=workdir) result, is_error, cost, limited_until = None, rc != 0, 0.0, None for line in lines: try: ev = json.loads(line) except ValueError: continue if ev.get("type") == "result": result, is_error, cost = ev.get("result") or "", bool(ev.get("is_error")), ev.get("total_cost_usd") or 0.0 if ev.get("type") == "rate_limit_event": info = ev.get("rate_limit_info") or ev # observed: status, rateLimitType, resetsAt if str(info.get("status", "")).lower() not in ("allowed", "allowed_warning", ""): limited_until = info.get("resetsAt") if limited_until or (is_error and LIMIT_RE.search(result or "")): cool_down("claude", until=limited_until) return "limit", result, cost return ("error" if is_error else "ok"), result, cost ``` Codex reads the prompt from stdin with `-`, prints progress to stderr and the final message to stdout (`-o` also writes it to a file): ```python async def run_codex(jid, job, preamble, workdir): args = ["codex", "exec", "--sandbox", "read-only" if job["mode"] == "ask" else "workspace-write", "--skip-git-repo-check", "--json", "-o", f"{workdir}/codex-last.md", "-C", workdir, "-"] rc, lines = await stream(jid, args, stdin_text=preamble + "\n\nTask:\n" + job["prompt"], cwd=workdir) if rc != 0 and LIMIT_RE.search("\n".join(lines[-30:])): cool_down("codex", minutes=20) return "limit", "\n".join(lines[-5:]), 0.0 return ("ok" if rc == 0 else "error"), read_text(f"{workdir}/codex-last.md"), 0.0 ``` The local model is an OpenAI-style call to Ollama or LiteLLM (`POST /v1/chat/completions`). Use it for drafts only. The fallback loop: ```python LIMIT_RE = re.compile(r"(usage limit|rate.?limit|limit reached|quota|at capacity|overloaded|too many requests)", re.I) async def run_job(jid): job = load_job(jid) order = [job["plan"]["engine"]] + [e for e in CFG["engine_order"] if e != job["plan"]["engine"]] for engine in order: if engine != order[0] and cooling_down(engine): continue status, text, cost = await ENGINES[engine](jid, job, preamble_for(job), job_dir(jid)) if status != "limit": break finish(jid, engine, status, text, cost) notify(f"job {jid} {status} via {engine}", click=f"https://hub.example.com/#job={jid}") ``` ### 3.5 A worker that can't vanish Python's documentation is explicit about `asyncio.create_task`: "The event loop only keeps weak references to tasks." Keep references, or the worker can disappear and jobs sit queued forever. ```python TASKS = set() def spawn(coro): t = asyncio.create_task(coro) TASKS.add(t) t.add_done_callback(TASKS.discard) return t async def worker(): while True: if running_count() < CFG["max_parallel"] and (jid := next_queued()): set_status(jid, "running") spawn(run_job(jid)) await asyncio.sleep(1) @app.on_event("startup") async def startup(): mark_running_as_interrupted() # only running jobs; queued jobs stay queued spawn(worker()) ``` ### 3.6 Alerts: job id and link only ```python import urllib.request def notify(text, click=None): if not CFG["ntfy_url"]: return headers = {"Title": "hub"} if click: headers["Click"] = click urllib.request.urlopen(urllib.request.Request(CFG["ntfy_url"], data=text.encode(), headers=headers), timeout=10) ``` Never put the prompt, the result or customer data in an alert. ntfy has no end-to-end encryption; the link opens the hub, which is behind the sign-in. ### 3.7 Running a job on another machine Prefix the engine command so it runs over SSH as a dedicated account on that machine: ```python prefix = ["ssh", "-i", "/home/you/.ssh/hub_runner", "-o", "BatchMode=yes", "runner@10.0.0.20"] ``` For a Windows machine: call `claude.cmd` and `codex.cmd` (PowerShell's default execution policy blocks npm's `.ps1` shims), and send the instructions on stdin, because multi-line arguments sent over SSH into PowerShell get mangled. ## 4. Run it and publish it A user service (enable lingering with `loginctl enable-linger you` so it runs without a login): ```ini # ~/.config/systemd/user/hub.service [Unit] Description=AI prompt hub [Service] WorkingDirectory=%h/hub ExecStart=%h/hub/.venv/bin/uvicorn server:app --app-dir %h/hub/app --host 127.0.0.1 --port 8790 --workers 1 Restart=always RestartSec=5 UMask=0077 [Install] WantedBy=default.target ``` Cloudflare: 1. Zero Trust, Access, Applications: add `hub.example.com` to an application whose policy allows only your accounts. Copy its AUD tag. 2. Networks, Tunnels, your tunnel, Routes: add a published application `hub.example.com` to `http://127.0.0.1:8790`, and turn on **Protect with Access** with your team name and that AUD tag. 3. Check from outside: `curl -s -o /dev/null -w "%{http_code} %{redirect_url}\n" https://hub.example.com/` must answer 302 to your Access login, for `/`, a health URL and any API path alike. ## 5. Clients **URL prompts.** `https://hub.example.com/?q=&target=server&mode=ask` opens a plan with a Run button. Make phone shortcuts and bookmarks from it. **Terminal client.** A short Python script with no dependencies that sends `X-Hub-Token` to `http://127.0.0.1:8790`, shows the plan, waits for Enter, then polls `/api/jobs//log?offset=N` until the job ends. **Desktop icons with a key that can do one thing.** On the hub server, in `~/.ssh/authorized_keys`: ``` restrict,pty,command="/usr/local/bin/hub" ssh-ed25519 AAAA... desktop-hub ``` The desktop's launcher runs `ssh -t -i you@server`. Test it: an interactive session opens the hub; sending `id; hostname` runs nothing. **A Windows account for agent jobs.** Elevated PowerShell on the Windows machine: ```powershell $b = New-Object byte[] 24; [Security.Cryptography.RandomNumberGenerator]::Create().GetBytes($b) $pw = ConvertTo-SecureString ([Convert]::ToBase64String($b) + 'a1!') -AsPlainText -Force New-LocalUser -Name runner -Password $pw -PasswordNeverExpires -Description 'hub job runner' Add-LocalGroupMember -Group Users -Member runner # never Administrators # Create the profile once (start a process as the user), then write C:\Users\runner\.ssh\authorized_keys # with ACLs for runner, SYSTEM and Administrators only, or OpenSSH ignores the file. ``` Then sign Claude Code and Codex in once as that account. ## 6. Remote Control and remote devices A job that needs a person can continue as a Claude Code Remote Control session: ```sh tmux new-session -d -s hub-JOB -c /path/to/job "claude --remote-control hub-JOB --permission-mode auto 'Continue job JOB. Read prompt.md and result.md first.'" ``` Pick it up in claude.ai/code or the Claude app. Remote Control needs a Pro, Max, Team or Enterprise plan; API keys are not supported. ## 7. Test before you trust it - A URL with `?q=` shows a plan and runs nothing until you press Run. - Replaying a Run request fails (the token is single use). - From outside, every path answers with the Access login redirect. - A forged `Cf-Access-Jwt-Assertion` (any made-up token) gets 401. - An ask-mode job that tries to write a file is refused. - Kill the hub mid-job: running jobs show interrupted, queued jobs still run after restart. - Force a Claude limit (or simulate one) and watch the job move to Codex with the reason in the log. - The alert on your phone contains a job id and link, nothing else. ## 8. Traps we hit - Loopback is not authentication when a tunnel connector shares the machine. - `asyncio.create_task` without a stored reference: the worker vanished. - A startup routine that marked queued jobs interrupted. - A 1.5B local model writing job titles ("AWS EC2 instance" for a one-line question). Use the prompt's first sentence. - A notes-lookup threshold set too low attached unrelated notes. Test with prompts that should match nothing. - PowerShell blocks npm's `.ps1` shims by default; multi-line SSH arguments into PowerShell break. ## Sources - Claude Code non-interactive mode: https://code.claude.com/docs/en/headless - Claude Code Remote Control: https://code.claude.com/docs/en/remote-control - Codex non-interactive mode: https://developers.openai.com/codex/noninteractive - Cloudflare Access, validate JWTs: https://developers.cloudflare.com/cloudflare-one/access-controls/applications/http-apps/authorization-cookie/validating-json/ - OpenSSH authorized_keys options: https://man.openbsd.org/sshd#AUTHORIZED_KEYS_FILE_FORMAT - Python asyncio.create_task: https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task - ntfy: https://docs.ntfy.sh/