diff --git a/.github/workflows/images.yml b/.github/workflows/images.yml index 0915b21..dd8fb90 100644 --- a/.github/workflows/images.yml +++ b/.github/workflows/images.yml @@ -16,6 +16,7 @@ on: - "hello-world/**" - "echo/**" - "tiles/**" + - "api-proxy/**" - ".github/workflows/images.yml" pull_request: paths: *image_paths @@ -34,7 +35,7 @@ jobs: strategy: fail-fast: false matrix: - example: [hello-world, echo, tiles] + example: [hello-world, echo, tiles, api-proxy] steps: - uses: actions/checkout@v7 diff --git a/README.md b/README.md index 1ee4423..f88075d 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ flowchart LR The orchestrator is a **transparent reverse proxy**: every endpoint you expose is passed through to your app unchanged, so you write an ordinary service and it runs on the network as-is. The transports supported today: -- **HTTP** request/response — the common case. (`hello-world`, `tiles`) +- **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`) - **HTTP + SSE** — streamed / token responses. (`vllm`) - **Trickle** — continuous realtime video in/out. (`echo`) - **WebSocket** — long-lived bidirectional sessions. (external: `scope`) @@ -37,12 +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 | -| [`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 (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 | 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. @@ -53,7 +54,7 @@ This set stays **minimal and curated**: it covers each value of the axes above ( How the app attaches to the orchestrator: - **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`) -- **Static** — the orchestrator is configured with the app's URL in a `runners.json` and health-polls it; the app needs no SDK. Best for fixed, long-running deployments. (`vllm`) +- **Static** — the orchestrator is configured with the app's URL in a `runners.json` and health-polls it; the app needs no SDK. Best for fixed, long-running deployments. (`vllm`, `api-proxy`) The arrow flips — dynamic, the app announces itself; static, the orchestrator is told about a passive app: @@ -74,7 +75,7 @@ flowchart LR Chosen _at_ registration (above); **defaults to `persistent`** — set on both `register_runner(...)` and in `runners.json`. The examples set it explicitly. - **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `vllm`) -- **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. (`hello-world`, `tiles`) +- **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. (`hello-world`, `tiles`, `api-proxy`) > [!NOTE] > The `vllm` example is single-shot by nature but stays **persistent** for now: it meters per second across a reserved session, and true per-token billing is brokerage for the gateway/signer layer. @@ -83,7 +84,7 @@ Chosen _at_ registration (above); **defaults to `persistent`** — set on both ` The client side depends on the runner's mode: -- **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`) +- **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`, `api-proxy`) - **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `vllm`) Each example's `client.py` shows its exact calls — grep `# Livepeer:` to find them. @@ -92,9 +93,10 @@ Each example's `client.py` shows its exact calls — grep `# Livepeer:` to find Apps that integrate the live runner and live in their own repos — production deployments and standalone examples alike. This table is links-only: the code, CI, and support stay with the author. -| Project | What it is | Transport | -| -------------------------------------------------------------------------- | ------------------------------------------------ | ------------------- | -| [daydreamlive/scope](https://github.com/daydreamlive/scope/tree/ja/runner) | Real-time AI video with downloadable LoRA models | WebSocket + trickle | +| Project | What it is | Transport | +| -------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- | ------------------- | +| [daydreamlive/scope](https://github.com/daydreamlive/scope/tree/ja/runner) | Real-time AI video with downloadable LoRA models | WebSocket + trickle | +| [livepeer/api-proxy](https://github.com/livepeer/api-proxy) | Attach several API endpoints dynamically — key storage and request stats for operators | HTTP | Built one? [Open a PR](https://github.com/livepeer/runner-app-examples/compare) that adds a row. To make your repo easy to find, follow the community convention: diff --git a/api-proxy/.env.example b/api-proxy/.env.example new file mode 100644 index 0000000..4a234f8 --- /dev/null +++ b/api-proxy/.env.example @@ -0,0 +1,31 @@ +# Copy to .env (gitignored) and fill in. Never commit secrets. +# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only. + +# Upstream credential the app injects (required, offchain too): +# huggingface.co → settings → tokens. +HF_TOKEN=hf_your_token + +# --- On-chain (paid) only below; offchain ignores these. --- + +NETWORK=arbitrum-one-mainnet +ETH_RPC_URL=https://arb1.arbitrum.io/rpc + +# Signer (payer): needs an on-chain deposit + reserve. +SIGNER_KEYSTORE_DIR=/absolute/path/to/signer-keystore +SIGNER_ETH_ACCT=0xYourSignerAddress +SIGNER_ETH_PASSWORD=your-signer-keystore-password + +# Orchestrator operating key (split-key): needs ETH for gas to redeem tickets. +ORCH_KEYSTORE_DIR=/absolute/path/to/operator-keystore +ORCH_ETH_ACCT=0xYourOperatorAddress +ORCH_ETH_PASSWORD=your-operator-keystore-password +# Registered orch = ticket recipient (-ethOrchAddr); empty = use the operating key. +ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator + +# The runner's price lives in runners.json (static runner): USD billed once per +# call (fixed pricing). Keep it under ~0.00019: the signer signs at most 100 +# tickets per payment, and the demo orchestrator runs -ticketEV=1e9. +# +# Signer's max-price cap (payer side), compared per billing unit. With fixed +# pricing the unit is one call, so this must exceed the runners.json price. +MAX_PRICE_PER_UNIT=0.000111USD diff --git a/api-proxy/.gitignore b/api-proxy/.gitignore new file mode 100644 index 0000000..ceb6b6b --- /dev/null +++ b/api-proxy/.gitignore @@ -0,0 +1,2 @@ +# Client output +api-proxy-out.jpg diff --git a/api-proxy/Dockerfile b/api-proxy/Dockerfile new file mode 100644 index 0000000..05eaef8 --- /dev/null +++ b/api-proxy/Dockerfile @@ -0,0 +1,15 @@ +# 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 new file mode 100644 index 0000000..668bf0e --- /dev/null +++ b/api-proxy/README.md @@ -0,0 +1,52 @@ +# API-proxy app (passthrough to an upstream API) + +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. + +| | | +| ------------ | ------------------------------------------ | +| App id | `livepeer-example/api-proxy` | +| Runner mode | single-shot | +| Registration | static (orchestrator config + health poll) | +| Transport | HTTP (JSON envelope in, JSON/base64 out) | +| 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. + +## 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. + +**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. + +> [!NOTE] +> Registration can also be **dynamic**: an operator tool can `register_runner` several API endpoints at runtime, each as its own priced capability, without touching the orchestrator config. See [livepeer/api-proxy](https://github.com/livepeer/api-proxy) for an example of dynamic endpoint registration, with key storage and request stats for orchestrator operators. + +## 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 +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`. + +## Run on-chain (paid) + +Layer `compose.onchain.yml` to run the orchestrator on-chain with a remote signer paying each call — one fixed payment per image, at the price `runners.json` advertises. For the required RPC and wallets see [On-chain (paid) setup](../README.md#on-chain-paid-setup) in the repo README. + +```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 +uv run client.py --prompt "a watercolor painting of a llama writing code" \ + --discovery https://localhost:8935/discovery \ + --signer http://localhost:7936 +docker compose -f compose.yml -f compose.onchain.yml down +``` + +Each call is one paid single-shot session — the orchestrator reserves it, takes one fixed payment, and releases it when the response returns. diff --git a/api-proxy/client.py b/api-proxy/client.py new file mode 100644 index 0000000..ba67a54 --- /dev/null +++ b/api-proxy/client.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""api-proxy client: discover a runner, proxy a text-to-image call, save the image. + +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. + +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. +""" + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import logging +from pathlib import Path + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.live_runner import call_runner +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" +DEFAULT_OUTPUT = "api-proxy-out.jpg" + +log = logging.getLogger("api-proxy-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Run the api-proxy Live Runner demo.") + 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( + "--signer", default="", help="Remote signer base URL (on-chain/paid path)." + ) + return parser.parse_args() + + +async def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + try: + cursor = await runner_selector( # Livepeer: 1 + discovery_url=args.discovery, app=APP_ID + ) + 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 + runner=runner, # discovery metadata tells call_runner the price unit + runner_url=runner.url.rstrip("/") + "/proxy", + payload=envelope, + signer_url=args.signer.strip() or None, + timeout=120.0, # a hosted diffusion model can take tens of seconds + ) + + 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) + except LivepeerGatewayError as exc: + raise SystemExit(f"ERROR: {exc}") from exc + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/api-proxy/compose.onchain.yml b/api-proxy/compose.onchain.yml new file mode 100644 index 0000000..8d9a957 --- /dev/null +++ b/api-proxy/compose.onchain.yml @@ -0,0 +1,50 @@ +# On-chain payment overlay for api-proxy. Layer it on the offchain base: +# docker compose -f compose.yml -f compose.onchain.yml up -d --build +# +# Adds the shared remote signer and re-points the orchestrator on-chain (see +# ../compose.onchain.yml), with runners.json mounted so the statically attached +# runner advertises its fixed price and the orchestrator issues a payment +# challenge. Requires a local .env (gitignored); copy .env.example and fill it +# in. Then pay through the signer: +# uv run client.py --prompt "..." \ +# --discovery https://localhost:8935/discovery \ +# --signer http://localhost:7936 + +services: + signer: + extends: + file: ../compose.onchain.yml + service: signer + ports: + - "7936:7936" + + orchestrator: + extends: + file: ../compose.onchain.yml + service: orchestrator + # Same flags as ../compose.onchain.yml, restated to add -liveRunnerConfig + # (`extends` can't append one) + mount runners.json. + command: + - -orchestrator + - -useLiveRunners + - -serviceAddr=127.0.0.1:8935 + - -httpAddr=0.0.0.0:8935 + - -liveRunnerAddr=https://orchestrator:8935 + - -orchSecret=abcdef + - -network=${NETWORK} + - -ethUrl=${ETH_RPC_URL} + - -ethKeystorePath=/keystore + - -ethAcctAddr=${ORCH_ETH_ACCT} + - -ethPassword=${ORCH_ETH_PASSWORD} + - -ethOrchAddr=${ORCH_ONCHAIN_ADDR} + # Price comes from the app's registration; -pricePerUnit only sets the + # unused legacy base price, but an on-chain O won't boot without it. + - -pricePerUnit=0 + - -ticketEV=1000000000 + - -maxFaceValue=4000000000000000 + - -liveRunnerConfig=/config/runners.json + - -monitor=false + - -v=6 + volumes: + - ${ORCH_KEYSTORE_DIR}:/keystore:ro + - ./runners.json:/config/runners.json:ro diff --git a/api-proxy/compose.yml b/api-proxy/compose.yml new file mode 100644 index 0000000..e838b5b --- /dev/null +++ b/api-proxy/compose.yml @@ -0,0 +1,40 @@ +# End-to-end offchain demo: orchestrator + api-proxy app, attached as a static +# runner via runners.json (-liveRunnerConfig). +# +# 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 +# uv run client.py --prompt "..." --discovery https://localhost:8935/discovery + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + # `extends` can't append a single flag to `command`, so the command is + # restated here with -liveRunnerConfig added and runners.json mounted in. + command: + - -orchestrator + - -useLiveRunners + - -serviceAddr=127.0.0.1:8935 + - -httpAddr=0.0.0.0:8935 + - -liveRunnerAddr=https://orchestrator:8935 + - -orchSecret=abcdef + - -network=offchain + - -monitor=false + - -liveRunnerConfig=/config/runners.json + - -v=6 + volumes: + - ./runners.json:/config/runners.json:ro + + app: + build: . + 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 diff --git a/api-proxy/pyproject.toml b/api-proxy/pyproject.toml new file mode 100644 index 0000000..24f4a2a --- /dev/null +++ b/api-proxy/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "livepeer-api-proxy" +version = "0.1.0" +description = "API proxy example app for the Livepeer network." +requires-python = ">=3.12" +dependencies = [ + "aiohttp", + "livepeer-gateway", +] + +# livepeer-gateway is not on PyPI yet; pull it from the branch. +[tool.uv.sources] +livepeer-gateway = { git = "https://github.com/livepeer/livepeer-python-gateway", branch = "ja/live-runner" } diff --git a/api-proxy/runner.py b/api-proxy/runner.py new file mode 100644 index 0000000..66392f0 --- /dev/null +++ b/api-proxy/runner.py @@ -0,0 +1,124 @@ +#!/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: + payload = await request.json() + 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 new file mode 100644 index 0000000..6907e49 --- /dev/null +++ b/api-proxy/runners.json @@ -0,0 +1,13 @@ +{ + "runners": [ + { + "label": "api-proxy", + "app": "livepeer-example/api-proxy", + "runner_url": "http://app:8989", + "health_url": "/health", + "mode": "single-shot", + "capacity": 1, + "price_info": { "price": 0.0001, "unit": "fixed" } + } + ] +}