diff --git a/.github/workflows/images.yml b/.github/workflows/images.yml index dd8fb90..0915b21 100644 --- a/.github/workflows/images.yml +++ b/.github/workflows/images.yml @@ -16,7 +16,6 @@ on: - "hello-world/**" - "echo/**" - "tiles/**" - - "api-proxy/**" - ".github/workflows/images.yml" pull_request: paths: *image_paths @@ -35,7 +34,7 @@ jobs: strategy: fail-fast: false matrix: - example: [hello-world, echo, tiles, api-proxy] + example: [hello-world, echo, tiles] steps: - uses: actions/checkout@v7 diff --git a/README.md b/README.md index f88075d..505213c 100644 --- a/README.md +++ b/README.md @@ -37,13 +37,13 @@ Need a schema that isn't here? [Open an issue](https://github.com/livepeer/runne ## Examples -| Example | Goal | Registration | Mode | Transport | Pricing | -| ------------------------------ | ------------------------------------------------------------------------------------- | ------------ | ---------------------------------- | -------------------- | ----------------- | -| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | single-shot | HTTP (JSON) | fixed | -| [`tiles`](./tiles) | Capacity fan-out — one call per tile | dynamic | single-shot | HTTP (base64 PNG) | fixed | -| [`api-proxy`](./api-proxy) | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static | single-shot | HTTP (JSON envelope) | fixed | -| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | — (offchain only) | -| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | persistent (single-shot by nature) | HTTP + SSE | hour | +| Example | Goal | Registration | Mode | Transport | Pricing | +| ------------------------------ | ------------------------------------------------------------------------------------- | ------------ | ---------------------------------- | ----------------- | ----------------- | +| [`hello-world`](./hello-world) | The simplest app: one request, one response | dynamic | single-shot | HTTP (JSON) | fixed | +| [`tiles`](./tiles) | Capacity fan-out — one call per tile | dynamic | single-shot | HTTP (base64 PNG) | fixed | +| [`api-proxy`](./api-proxy) | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static | single-shot | HTTP (JPEG bytes) | fixed | +| [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | — (offchain only) | +| [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | persistent (single-shot by nature) | HTTP + SSE | hour | Start with `hello-world` (the smallest end-to-end path); the others each layer on one new idea. More will follow, including a full example that exercises every feature. Each is self-contained and runs **offchain** (free, no wallet); most also run **on-chain** (paid) — see each README. diff --git a/api-proxy/Dockerfile b/api-proxy/Dockerfile deleted file mode 100644 index 05eaef8..0000000 --- a/api-proxy/Dockerfile +++ /dev/null @@ -1,15 +0,0 @@ -# api-proxy example app (http server). No Livepeer code — the orchestrator -# attaches it statically via runners.json. -FROM python:3.12-slim - -# Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`. -ENV PYTHONUNBUFFERED=1 - -RUN pip install --no-cache-dir aiohttp - -WORKDIR /app -COPY runner.py ./ - -EXPOSE 8989 - -ENTRYPOINT ["python", "runner.py"] diff --git a/api-proxy/README.md b/api-proxy/README.md index 6bbe6ac..8ebb83d 100644 --- a/api-proxy/README.md +++ b/api-proxy/README.md @@ -1,25 +1,29 @@ -# API-proxy app (passthrough to an upstream API) +# API-proxy app (a runner that is pure config) -The live runner can also **pass calls through to an API that runs somewhere else** — a hosted model API, a SaaS endpoint, a service on your own infrastructure. The orchestrator operator attaches the proxy as a runner — **statically or dynamically; this example uses static** — and offers the upstream as a paid capability on the network: `POST /proxy` forwards a JSON envelope to the upstream and returns its response. The demo upstream is the **Hugging Face text-to-image inference API** ([Stable Diffusion 3 medium](https://huggingface.co/stabilityai/stable-diffusion-3-medium-diffusers) by default), but `--upstream` points it at any REST API. +The live runner can also **pass calls through to an API that runs somewhere else** — here the **Hugging Face text-to-image inference API**. This example's runner is a **stock nginx**: [nginx.conf.template](nginx.conf.template) forwards each call to one pinned model URL and injects the operator's token. There is **no app code at all** — the orchestrator operator offers a hosted model ([Stable Diffusion 3 medium](https://huggingface.co/stabilityai/stable-diffusion-3-medium-diffusers) by default) as a paid capability with two config files. -| | | -| ------------ | ------------------------------------------ | -| App id | `livepeer-example/api-proxy` | -| Runner mode | single-shot | -| Registration | static (orchestrator config + health poll) | -| Transport | HTTP (JSON envelope in, JSON/base64 out) | -| Pricing | fixed (one price per call) | -| Port | 8989 | +| | | +| ------------ | -------------------------------------------- | +| App id | `livepeer-example/stable-diffusion-3-medium` | +| Runner mode | single-shot | +| Registration | static (orchestrator config + health poll) | +| Transport | HTTP (HF payload in, JPEG bytes out) | +| Pricing | fixed (one price per call) | +| Port | 8989 | Prerequisites (Docker, `uv`, and the not-yet-released `livepeer-gateway` SDK — pinned in `pyproject.toml`) and the shared on-chain/payment setup live in the [repo README](../README.md). The demo upstream additionally needs a **Hugging Face API token** (`HF_TOKEN`, from [huggingface.co → settings → tokens](https://huggingface.co/settings/tokens)) with inference-provider credits. ## How it's wired -The app is attached as a **static runner**: the orchestrator reads [runners.json](runners.json) via `-liveRunnerConfig` — app id, runner URL, single-shot mode, and the fixed price — and health-polls `/health`. There is **no Livepeer code in the app**: [runner.py](runner.py) is a plain aiohttp service. Each `/proxy` call forwards `{"method", "path", "headers", "json"}` to `/` and returns `{"status", "headers", "body"}` for text upstream bodies or `{"status", "headers", "body_b64"}` for binary ones (a generated image, say). The client calls it with `runner_selector` → `call_runner` ([client.py](client.py)) — discover, then one **single-shot** call per request; the orchestrator reserves a session per call and releases it when the response returns. Grep `# Livepeer:` in client.py to see the exact calls. +The app is attached as a **static runner**: the orchestrator reads [runners.json](runners.json) via `-liveRunnerConfig` — app id, runner URL, single-shot mode, and the fixed price — and health-polls `/health` (an nginx `return 200`). The `/proxy` location proxies to the pinned model URL (`MODEL` in [compose.yml](compose.yml)) with `Authorization: Bearer ` added. The caller's body is the [Hugging Face text-to-image payload](https://huggingface.co/docs/inference-providers/tasks/text-to-image) forwarded verbatim — `{"inputs": ""}` — and the image comes back as **raw JPEG bytes**. The client calls it with `runner_selector` → `call_runner(..., stream=True)` ([client.py](client.py)) — discover, then one **single-shot** call per image, reading the bytes with `aiter_bytes()`; the orchestrator reserves a session per call and releases it when the response returns. Grep `# Livepeer:` in client.py to see the exact calls. ## Offering an API as a capability — what this shows -Everything the operator sets lives on the operator's side. `runners.json` names the capability, the proxy's URL, and the **fixed per-call price**; the upstream credential (`UPSTREAM_TOKEN`, fed from `HF_TOKEN` by the compose files) sits in the app's environment, and the app injects it as a Bearer header on every forward — any `Authorization` a caller sends is dropped. Callers need no API key of their own: they discover the capability and pay **per call through Livepeer**, while the operator pays the upstream and prices above the per-call upstream cost. +Everything is operator-side config. `runners.json` names the capability and sets the **fixed per-image price**; the nginx config pins the model URL and holds the credential (`HF_TOKEN`, from `.env`). The pinned URL is also the security model: the operator's credential can only be spent on exactly the offered model. The config pins the method to `POST` and drops the caller's query string too, so the body is the only thing a caller controls: they choose nothing but the prompt, and never see an API key. They discover the capability and pay **per image through Livepeer**, while the operator pays the upstream and prices above the per-image upstream cost. + +The app id names the model, not the proxy, because that is what callers discover: they match it exactly, so it has to say what they get. Swapping `MODEL` means renaming the app id with it. + +Offering a second model is more config, not code: one more `runners.json` entry (its own app id and price) plus one more nginx service with a different `MODEL`. **Fixed pricing** is the natural fit: one call is one bounded unit of work, so the runner bills one flat price per call instead of metering time. @@ -29,13 +33,14 @@ Everything the operator sets lives on the operator's side. `runners.json` names ## Run offchain (free) ```sh -HF_TOKEN=hf_... docker compose up -d --build -curl -sk https://localhost:8935/discovery | jq '.[].runners[].app' # confirm livepeer-example/api-proxy registered +cp .env.example .env # fill in HF_TOKEN; ignore the on-chain block +docker compose up -d +curl -sk https://localhost:8935/discovery | jq '.[].runners[].app' # confirm livepeer-example/stable-diffusion-3-medium registered uv run client.py --prompt "a watercolor painting of a llama writing code" docker compose down ``` -`compose.yml` brings up an orchestrator (`-useLiveRunners -liveRunnerConfig`) and the app (proxying `https://router.huggingface.co`). The client builds the envelope for one text-to-image call, sends it through the orchestrator, and writes `api-proxy-out.jpg`. +`compose.yml` brings up an orchestrator (`-useLiveRunners -liveRunnerConfig`) and the nginx runner. The client sends one prompt through the orchestrator and writes `api-proxy-out.jpg`. ## Run on-chain (paid) @@ -43,7 +48,7 @@ Layer `compose.onchain.yml` to run the orchestrator on-chain with a remote signe ```sh cp .env.example .env # fill in HF_TOKEN, RPC, network, keystore paths, accounts -docker compose -f compose.yml -f compose.onchain.yml up -d --build +docker compose -f compose.yml -f compose.onchain.yml up -d uv run client.py --prompt "a watercolor painting of a llama writing code" \ --discovery https://localhost:8935/discovery \ --signer http://localhost:7936 diff --git a/api-proxy/client.py b/api-proxy/client.py index ba67a54..3941679 100644 --- a/api-proxy/client.py +++ b/api-proxy/client.py @@ -1,24 +1,22 @@ #!/usr/bin/env python3 -"""api-proxy client: discover a runner, proxy a text-to-image call, save the image. +"""api-proxy client: discover a runner, send a prompt, stream the image back. -Builds the generic /proxy envelope for one concrete upstream — the Hugging Face -text-to-image inference API — and decodes the binary response. Any other REST -API is the same envelope with a different method/path/json. +The request body is the Hugging Face text-to-image payload as-is +({"inputs": ""}); the runner forwards it verbatim and the image comes +back as raw JPEG bytes, received with call_runner's streaming mode. Livepeer integration (grep `# Livepeer:`): - 1. runner_selector() — discover orchestrators advertising the app - 2. call_runner() — call the app through the orchestrator; on the paid path it - answers the 402 payment challenge inline (one fixed payment - per call). Single-shot needs no reserve/stop: the - orchestrator reserves the session for this one request and - releases it when the response returns. + 1. runner_selector() — discover orchestrators advertising the app + 2. call_runner(stream=True) — call the app through the orchestrator and read the + raw response bytes; on the paid path it answers the + 402 payment challenge inline (one fixed payment per + image). Single-shot needs no reserve/stop. """ from __future__ import annotations import argparse import asyncio -import base64 import logging from pathlib import Path @@ -27,8 +25,7 @@ from livepeer_gateway.selection import runner_selector DEFAULT_DISCOVERY = "https://localhost:8935/discovery" -APP_ID = "livepeer-example/api-proxy" -DEFAULT_MODEL = "stabilityai/stable-diffusion-3-medium-diffusers" +APP_ID = "livepeer-example/stable-diffusion-3-medium" DEFAULT_OUTPUT = "api-proxy-out.jpg" log = logging.getLogger("api-proxy-client") @@ -39,11 +36,6 @@ def _parse_args() -> argparse.Namespace: parser.add_argument( "--prompt", default="a watercolor painting of a llama writing code" ) - parser.add_argument( - "--model", - default=DEFAULT_MODEL, - help="Hugging Face text-to-image model (the upstream path).", - ) parser.add_argument("--output", default=DEFAULT_OUTPUT, help="output image path") parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) parser.add_argument( @@ -64,31 +56,21 @@ async def main() -> None: runner = cursor.candidates[0] log.info("app_url=%s", runner.url) - # The envelope the app forwards upstream: here a Hugging Face - # text-to-image call, but any method/path/json works. - envelope = { - "method": "POST", - "path": f"/hf-inference/models/{args.model}", - "json": {"inputs": args.prompt}, - } - result = await call_runner( # Livepeer: 2 + # NOTE: Streamed because the buffered path still assumes JSON; once + # livepeer-python-gateway#51 lands it returns result.raw. + stream = await call_runner( # Livepeer: 2 runner=runner, # discovery metadata tells call_runner the price unit runner_url=runner.url.rstrip("/") + "/proxy", - payload=envelope, + payload={"inputs": args.prompt}, # the HF payload, forwarded as-is signer_url=args.signer.strip() or None, - timeout=120.0, # a hosted diffusion model can take tens of seconds + stream=True, # the image comes back as raw bytes, not JSON ) + async with stream: + image = b"".join([chunk async for chunk in stream.aiter_bytes()]) - status = result.data.get("status") - if status != 200: - body = result.data.get("body") or result.data.get("error") - raise LivepeerGatewayError(f"upstream returned {status}: {body}") - b64 = result.data.get("body_b64") - if not isinstance(b64, str) or not b64: - raise LivepeerGatewayError("upstream response was not binary (no body_b64)") out_path = Path(args.output).expanduser() - out_path.write_bytes(base64.b64decode(b64)) - log.info("wrote %s", out_path) + out_path.write_bytes(image) + log.info("wrote %s (%d bytes, %s)", out_path, len(image), stream.content_type) except LivepeerGatewayError as exc: raise SystemExit(f"ERROR: {exc}") from exc diff --git a/api-proxy/compose.yml b/api-proxy/compose.yml index e838b5b..420a937 100644 --- a/api-proxy/compose.yml +++ b/api-proxy/compose.yml @@ -1,11 +1,10 @@ -# End-to-end offchain demo: orchestrator + api-proxy app, attached as a static -# runner via runners.json (-liveRunnerConfig). +# End-to-end offchain demo: orchestrator + the api-proxy "app" — a stock nginx +# attached as a static runner via runners.json (-liveRunnerConfig). There is no +# app code at all: nginx.conf.template is the whole runner. # -# The orchestrator service is defined once in ../compose.orchestrator.yml and -# pulled in with `extends`. Needs HF_TOKEN (the upstream credential the app -# injects) in a local .env — copy .env.example. Once up, call it from the host -# with the SDK: -# docker compose up -d --build +# Needs HF_TOKEN (the upstream credential nginx injects) in a local .env — copy +# .env.example. Once up, call it from the host with the SDK: +# docker compose up -d # uv run client.py --prompt "..." --discovery https://localhost:8935/discovery services: @@ -30,11 +29,13 @@ services: - ./runners.json:/config/runners.json:ro app: - build: . + image: nginx:1.27-alpine container_name: example_apps_api_proxy environment: # The operator-held upstream credential (huggingface.co → settings → tokens). - - UPSTREAM_TOKEN=${HF_TOKEN:?set HF_TOKEN in .env (copy .env.example)} - command: - - --host=0.0.0.0 - - --upstream=https://router.huggingface.co + - HF_TOKEN=${HF_TOKEN:?set HF_TOKEN in .env (copy .env.example)} + # The one model this runner offers; swap it here, and rename the app id + # in runners.json to match (the app id is what callers discover). + - MODEL=stabilityai/stable-diffusion-3-medium-diffusers + volumes: + - ./nginx.conf.template:/etc/nginx/templates/default.conf.template:ro diff --git a/api-proxy/nginx.conf.template b/api-proxy/nginx.conf.template new file mode 100644 index 0000000..f6498ac --- /dev/null +++ b/api-proxy/nginx.conf.template @@ -0,0 +1,29 @@ +# The whole runner: a stock nginx forwards each call to one pinned Hugging Face +# model URL and injects the operator's token. The orchestrator health-polls +# /health and reverse-proxies caller requests to /proxy. HF_TOKEN and MODEL are +# substituted from the environment by the nginx image's template mechanism. +server { + listen 8989; + + location = /health { + default_type application/json; + return 200 '{"status":"ok"}'; + } + + # The pinned URL is the security model: the operator's credential can only + # be spent on exactly this model, callers choose nothing but the payload. + # Method and query string are part of "nothing": nginx would forward both + # upstream as the caller sent them, so pin the method and drop the args. + location = /proxy { + limit_except POST { deny all; } + set $args ''; + proxy_pass https://router.huggingface.co/hf-inference/models/${MODEL}; + proxy_set_header Authorization "Bearer ${HF_TOKEN}"; + # Hugging Face content-negotiates: with the SDK's Accept: + # application/json it answers with a base64 PNG in a JSON string + # instead of the image. Ask for the image the runner promises. + proxy_set_header Accept "image/jpeg"; + proxy_ssl_server_name on; + proxy_read_timeout 120s; # a hosted diffusion model can take tens of seconds + } +} diff --git a/api-proxy/runner.py b/api-proxy/runner.py deleted file mode 100644 index 7e92214..0000000 --- a/api-proxy/runner.py +++ /dev/null @@ -1,131 +0,0 @@ -#!/usr/bin/env python3 -"""api-proxy app: pass calls through to an upstream HTTP API, offered on Livepeer. - -The upstream capability runs somewhere else; this app is the passthrough the -orchestrator operator offers on the network. It is attached as a static runner -(runners.json → -liveRunnerConfig), so there is no Livepeer code here at all: -the orchestrator health-polls /health and reverse-proxies calls to /proxy. - -`POST /proxy` takes a JSON envelope describing the upstream call -({"method", "path", "headers", "json"}) and returns the upstream response -({"status", "headers", "body"} for text, {"status", "headers", "body_b64"} for -binary). The upstream credential stays server-side: set UPSTREAM_TOKEN and the -app injects it as a Bearer token on every forward — callers pay Livepeer per -call and never see an API key. -""" - -from __future__ import annotations - -import argparse -import base64 -import logging -import os -from contextlib import suppress - -import aiohttp -from aiohttp import web - -DEFAULT_HOST = "127.0.0.1" -DEFAULT_PORT = 8989 -DEFAULT_UPSTREAM = "https://router.huggingface.co" -UPSTREAM_TIMEOUT = 120 # a hosted diffusion model can take tens of seconds -_TEXT_TYPES = ("text/", "application/json", "application/xml", "application/x-ndjson") - -log = logging.getLogger("api-proxy") - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Live Runner api-proxy demo.") - parser.add_argument( - "--host", default=DEFAULT_HOST, help="Bind address (use 0.0.0.0 in containers)." - ) - parser.add_argument( - "--upstream", - default=DEFAULT_UPSTREAM, - help="Base URL of the API to proxy requests to.", - ) - return parser.parse_args() - - -async def _handle_health(request: web.Request) -> web.Response: - # The orchestrator health-polls this (health_url in runners.json); routing - # starts once it returns 200. - return web.json_response({"status": "ok"}) - - -async def _handle_proxy(request: web.Request) -> web.Response: - try: - payload = await request.json() - except Exception: - payload = None - if not isinstance(payload, dict) or not isinstance( - payload.get("headers") or {}, dict - ): - raise web.HTTPBadRequest(text="body must be a JSON envelope object") - method = str(payload.get("method", "GET")).upper() - path = str(payload.get("path", "/")) - body = payload.get("json") - - # The operator's credential, not the caller's: drop any Authorization the - # caller sent and inject the stored token instead. - headers = { - k: v - for k, v in (payload.get("headers") or {}).items() - if k.lower() != "authorization" - } - token = request.app["token"] - if token: - headers["Authorization"] = f"Bearer {token}" - - upstream = request.app["upstream"].rstrip("/") + "/" + path.lstrip("/") - log.info("proxy %s %s", method, upstream) - - session: aiohttp.ClientSession = request.app["session"] - try: - async with session.request( - method, - upstream, - headers=headers, - json=body, - timeout=aiohttp.ClientTimeout(total=UPSTREAM_TIMEOUT), - ) as resp: - content_type = resp.headers.get("Content-Type", "") - raw = await resp.read() - except aiohttp.ClientError as exc: - return web.json_response({"error": str(exc)}, status=502) - - # Text-ish upstream bodies pass through as a string; anything else (e.g. a - # generated image) is base64 so the envelope stays plain JSON. - result: dict[str, object] = {"status": resp.status, "headers": dict(resp.headers)} - if content_type.startswith(_TEXT_TYPES) or "+json" in content_type: - result["body"] = raw.decode(errors="replace") - else: - result["body_b64"] = base64.b64encode(raw).decode() - return web.json_response(result) - - -def main() -> None: - logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" - ) - args = _parse_args() - - async def _on_startup(app: web.Application) -> None: - app["session"] = aiohttp.ClientSession() - - async def _on_cleanup(app: web.Application) -> None: - with suppress(Exception): - await app["session"].close() - - app = web.Application() - app["upstream"] = args.upstream - app["token"] = os.environ.get("UPSTREAM_TOKEN", "") - app.router.add_get("/health", _handle_health) - app.router.add_post("/proxy", _handle_proxy) - app.on_startup.append(_on_startup) - app.on_cleanup.append(_on_cleanup) - web.run_app(app, host=args.host, port=DEFAULT_PORT) - - -if __name__ == "__main__": - main() diff --git a/api-proxy/runners.json b/api-proxy/runners.json index 6907e49..7f52f63 100644 --- a/api-proxy/runners.json +++ b/api-proxy/runners.json @@ -2,7 +2,7 @@ "runners": [ { "label": "api-proxy", - "app": "livepeer-example/api-proxy", + "app": "livepeer-example/stable-diffusion-3-medium", "runner_url": "http://app:8989", "health_url": "/health", "mode": "single-shot",