From 323c96459c7a0b21b87290b17ada1b34aa68a1da Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 13:37:55 +0200 Subject: [PATCH 1/5] feat(live-runner): payment plumbing for single-shot ongoing payments - hoist _post_empty into http.py as post_empty (TCPConnector(ssl=False), typed status errors via the shared error mapper) so payment callers can branch on 404/409/403/482 - route LivePaymentSession.send_payment through it; the new payment_url kwarg targets a specific payment endpoint such as the session-scoped one - parse payment_interval_ms from payment challenges and surface it as server_payment_interval on call results and streams - carry the challenge session_id and the fixed-price payment_session drop through call_runner(stream=True) Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/http.py | 44 +++++++++++++++++ src/livepeer_gateway/live_runner.py | 71 ++++++++++++++++----------- src/livepeer_gateway/remote_signer.py | 58 ++++++++++------------ 3 files changed, 112 insertions(+), 61 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..db4057b 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -371,6 +371,50 @@ async def get_json( return await request_json(url, headers=headers, timeout=timeout) +async def post_empty( + url: str, + *, + headers: Optional[dict[str, str]] = None, + timeout: float = 5.0, +) -> None: + """ + POST an empty body to `url` and discard the response. + + Certificate verification is disabled, matching every other HTTP helper in + the SDK. Error responses raise the same typed errors as request_json + (SignerRefreshRequired for 480, SkipPaymentCycle for 482, LivepeerHTTPError + otherwise) so callers can branch on status codes. + """ + try: + client_timeout = aiohttp.ClientTimeout(total=timeout) + connector = aiohttp.TCPConnector(ssl=False) + async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: + async with session.post(url, data=b"", headers=headers) as resp: + if resp.status >= 400: + raw = await resp.text() + _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) + await resp.read() + except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError): + raise + except ConnectionRefusedError as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + except getattr(aiohttp, "ClientConnectorError", ()) as e: + os_error = getattr(e, "os_error", None) + if isinstance(os_error, ConnectionRefusedError): + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + + def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult: """ Normalize a URL for HTTP(S) endpoints. diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index fbe3010..93bd10c 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -29,7 +29,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .http import open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -129,6 +129,9 @@ class LiveRunnerCallResult: repr=False, compare=False, ) + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None @dataclass @@ -145,8 +148,15 @@ class LiveRunnerCallStream: runner_url: str runner: Optional[LiveRunnerInstance] payment_session: Optional[LivePaymentSession] - _session: aiohttp.ClientSession = field(repr=False, compare=False) - _response: aiohttp.ClientResponse = field(repr=False, compare=False) + # Session backing this single-shot request, from the payment challenge's + # manifest_id. Empty offchain: a stream body cannot supply the JSON-path + # session_id fallback. + session_id: str = "" + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None + _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True) + _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True) @property def content_type(self) -> str: @@ -296,10 +306,10 @@ async def close(self) -> None: _LOG.warning("Skipping live runner unregister without heartbeat secret") return try: - await _post_empty( + await post_empty( _join_endpoint(self.orchestrator_url, f"/runners/{quote(self.runner_id, safe='')}/unregister"), - {"Authorization": secret}, - self._timeout, + headers={"Authorization": secret}, + timeout=self._timeout, ) except Exception: _LOG.debug("Live runner unregister failed", exc_info=True) @@ -797,7 +807,17 @@ async def call_runner( headers=request_headers or None, ) return LiveRunnerCallStream( - resp.status, resp.headers, runner_url, runner, payment_session, session, resp, + status=resp.status, + headers=resp.headers, + runner_url=runner_url, + runner=runner, + payment_session=None if payment_type == "fixed" else payment_session, + session_id=session_id, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), + _session=session, + _response=resp, ) data = await request_json( @@ -819,6 +839,9 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -836,6 +859,9 @@ class _RunnerPaymentChallenge: payment_params: str orchestrator_url: str manifest_id: str + # Orchestrator debit cadence in seconds (payment_interval_ms); None when + # the orchestrator does not report it. + payment_interval_s: Optional[float] = None def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentChallenge: @@ -856,10 +882,16 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC if not isinstance(manifest_id, str) or not manifest_id: raise LivepeerGatewayError("Live runner payment challenge missing manifest_id") + interval_ms = data.get("payment_interval_ms") + payment_interval_s: Optional[float] = None + if isinstance(interval_ms, (int, float)) and not isinstance(interval_ms, bool) and interval_ms > 0: + payment_interval_s = float(interval_ms) / 1000.0 + return _RunnerPaymentChallenge( payment_params=payment_params, orchestrator_url=orchestrator_url, manifest_id=manifest_id, + payment_interval_s=payment_interval_s, ) @@ -977,10 +1009,10 @@ async def stop_runner_session( url = _join_endpoint(control_url, "stop") if isinstance(token, str) and token.strip(): request_headers = {"Livepeer-Session-Token": token} - await _post_empty( + await post_empty( url, - request_headers, - timeout, + headers=request_headers, + timeout=timeout, ) @@ -1146,25 +1178,6 @@ def _is_trickle_channel_response(value: object) -> bool: ) and ("internal_url" not in value or isinstance(value.get("internal_url"), str)) -async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: - try: - client_timeout = aiohttp.ClientTimeout(total=timeout) - connector = aiohttp.TCPConnector(ssl=False) - async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: - async with session.post(url, data=b"", headers=headers) as resp: - body = await resp.text() - if resp.status >= 400: - raise LivepeerGatewayError( - f"HTTP empty POST error: HTTP {resp.status}; body={body!r}" - ) - except LivepeerGatewayError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - - def _detect_gpu_pynvml() -> Optional[LiveRunnerGPU]: try: import pynvml # type: ignore[import-not-found] diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index fc94463..72aa598 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import base64 import json import logging @@ -12,8 +11,6 @@ from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen -import aiohttp - from . import lp_rpc_pb2 from .async_cache import async_lru_cache from .errors import LivepeerGatewayError, PaymentError, SignerRefreshRequired @@ -240,44 +237,41 @@ async def get_payment(self) -> GetPaymentResponse: await self._refresh_payment_params(orchestrator_url) attempts += 1 - async def send_payment(self, orchestrator_url: Optional[str] = None) -> None: + async def send_payment( + self, + orchestrator_url: Optional[str] = None, + *, + payment_url: Optional[str] = None, + ) -> None: + """Generate a payment and POST it to the orchestrator. + + ``payment_url`` targets a specific payment endpoint, such as the + session-scoped ``…/session/{session_id}/payment`` which 404s once the + session is released. Without it the payment goes to the orchestrator's + generic ``/payment`` endpoint, which credits the payer balance blindly. + + Raises LivepeerHTTPError on error responses so callers can branch on + the status code, and SkipPaymentCycle when the signer gates the cycle. + """ if not self._signer_url: return - target = orchestrator_url or self._orchestrator_url - if not target: - raise PaymentError("orchestrator_url is required before sending payment") + from .http import _http_origin, post_empty - from .http import _extract_error_message_from_body, _http_origin + if payment_url: + url = payment_url + else: + target = orchestrator_url or self._orchestrator_url + if not target: + raise PaymentError("orchestrator_url is required before sending payment") + url = f"{_http_origin(target)}/payment" payment = await self.get_payment() - url = f"{_http_origin(target)}/payment" headers = { "Livepeer-Payment": payment.payment, - "Livepeer-Segment": payment.seg_creds, + "Livepeer-Segment": payment.seg_creds or "", } - try: - timeout = aiohttp.ClientTimeout(total=5.0) - async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.post(url, data=b"", headers=headers) as resp: - if resp.status >= 400: - body = await resp.text() - message = _extract_error_message_from_body(body) - body_part = f"; body={message!r}" if message else "" - raise PaymentError( - f"HTTP payment error: HTTP {resp.status} from endpoint (url={url}){body_part}" - ) - await resp.read() - except PaymentError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e + await post_empty(url, headers=headers, timeout=5.0) async def _payment_request(self) -> GetPaymentResponse: from .http import _http_origin, post_json From 3936c771454bc6284217aa9ba8897aafdb5ccd24 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 13:39:17 +0200 Subject: [PATCH 2/5] feat(live-runner): fund metered single-shot calls while they run The orchestrator debits the prepaid balance on -livePaymentInterval while a metered (hour/720p) single-shot request is in flight and cancels it on the first under-funded tick; the signer's preroll only covers ~10s. Spawn a payment loop inside call_runner for the lifetime of the awaited request: first payment after one interval, cadence at 60% of the challenge's payment_interval_ms with a 3s fallback, stop on 404 (released) / 409 (fixed-price) / 403 (mismatch), skip on 482, retry transient errors. No opt-out flag: offchain mode is the no-payments path and fixed pricing never starts the loop. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 98 ++++++++++++++++++++++++++--- 1 file changed, 91 insertions(+), 7 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 93bd10c..3293ad7 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import contextlib import inspect import json import logging @@ -28,7 +29,12 @@ import aiohttp from .channel_reader import ChannelReader -from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired +from .errors import ( + LivepeerGatewayError, + LivepeerHTTPError, + SignerRefreshRequired, + SkipPaymentCycle, +) from .http import open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, @@ -49,6 +55,16 @@ "720p-pixel-seconds": "lv2v", "fixed": "fixed", } +# Metered payment types need ongoing funding while a call is in flight; the +# orchestrator debits the prepaid balance on -livePaymentInterval (5s default) +# and cancels the request on the first under-funded tick. +_METERED_PAYMENT_TYPES = frozenset({"live", "lv2v"}) +# Fallback client payment cadence when the challenge does not advertise +# payment_interval_ms (go-livepeer master does not, yet). +_DEFAULT_PAYMENT_INTERVAL_S = 3.0 +# Pay at a fraction of the server debit tick so a payment always lands +# before the next debit. +_PAYMENT_CADENCE_FRACTION = 0.6 # golang format duration, eg "10s" _DURATION_RE = re.compile(r"^\s*(?P[0-9]+(?:\.[0-9]+)?)(?Pns|us|\u00b5s|ms|s|m|h)\s*$") @@ -820,12 +836,31 @@ async def call_runner( _response=resp, ) - data = await request_json( - runner_url, - method=method, - payload=request_payload, - **request_kwargs, - ) + # Metered pricing: the orchestrator debits the prepaid balance + # while the request runs, so keep funding it for as long as we + # are waiting on the response. + pay_task: Optional[asyncio.Task[None]] = None + if payment_session is not None and payment_type in _METERED_PAYMENT_TYPES: + pay_task = asyncio.create_task( + _run_call_payments( + payment_session, + interval_s=_payment_cadence_s( + challenge.payment_interval_s if challenge is not None else None + ), + ) + ) + try: + data = await request_json( + runner_url, + method=method, + payload=request_payload, + **request_kwargs, + ) + finally: + if pay_task is not None: + pay_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await pay_task if not isinstance(data, dict): raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" @@ -895,6 +930,55 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) +def _payment_cadence_s(server_interval_s: Optional[float]) -> float: + if server_interval_s is not None and server_interval_s > 0: + return server_interval_s * _PAYMENT_CADENCE_FRACTION + return _DEFAULT_PAYMENT_INTERVAL_S + + +async def _run_call_payments( + payment_session: LivePaymentSession, + *, + interval_s: float, + payment_url: str = "", + on_released: Optional[Callable[[], None]] = None, +) -> None: + """Fund an in-flight metered single-shot call until cancelled. + + The 402 challenge payment already carried the signer's preroll, so the + first follow-up payment waits one interval. Runs until cancelled by the + caller or until the orchestrator reports a terminal state: 404 (session + released), 409 (fixed-price, nothing to fund), or 403 (session/payment + mismatch). Transient errors are retried on the next tick — the + orchestrator cancels the call itself if funding actually stops. + """ + while True: + await asyncio.sleep(interval_s) + try: + await payment_session.send_payment(payment_url=payment_url or None) + except asyncio.CancelledError: + raise + except SkipPaymentCycle: + _LOG.debug("Live runner call payment: signer skipped this cycle") + except LivepeerHTTPError as e: + if e.status_code == 404: + _LOG.info("Live runner call payment: session released; stopping payments") + if on_released is not None: + on_released() + return + if e.status_code == 409: + _LOG.debug("Live runner call payment: fixed-price session; stopping payments") + return + if e.status_code == 403: + _LOG.error( + "Live runner call payment: session/payment mismatch; stopping payments: %s", e + ) + return + _LOG.warning("Live runner call payment failed; retrying next cycle: %s", e) + except Exception as e: + _LOG.warning("Live runner call payment failed; retrying next cycle: %s", e) + + async def _get_runner_payment( challenge: _RunnerPaymentChallenge, *, From e1c6f8f7c257a12d72dc39c824059a9b0790574c Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 13:39:55 +0200 Subject: [PATCH 3/5] feat(live-runner): stream owns the payment loop for metered single-shot A streamed single-shot response outlives the call_runner call, so the LiveRunnerCallStream owns its funding: the loop starts with the stream, aclose() cancels it before tearing down the transport, and a 404 from the payment endpoint sets a read-only released flag. No public start/stop controls - a single-shot stream is its request, so stopping funding without closing is not a meaningful state. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 33 ++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 3293ad7..d9e312d 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -171,8 +171,15 @@ class LiveRunnerCallStream: # Orchestrator debit cadence in seconds, from the payment challenge's # payment_interval_ms. None when the orchestrator does not report it. server_payment_interval: Optional[float] = None + # True once the orchestrator reported the backing session gone (payment + # loop hit a 404). The stream itself ends when the orchestrator cancels + # the proxied request. + released: bool = False _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True) _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True) + _payment_task: Optional[asyncio.Task[None]] = field( + default=None, repr=False, compare=False, kw_only=True + ) @property def content_type(self) -> str: @@ -186,7 +193,18 @@ async def aiter_lines(self) -> AsyncIterator[str]: async for line in self._response.content: yield line.decode(errors="replace").rstrip("\n") + def _mark_released(self) -> None: + self.released = True + async def aclose(self) -> None: + # Stop funding before tearing down the transport so no payment is + # minted for a stream we are abandoning. + task = self._payment_task + self._payment_task = None + if task is not None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task self._response.release() await self._session.close() @@ -822,7 +840,7 @@ async def call_runner( payload=request_payload, headers=request_headers or None, ) - return LiveRunnerCallStream( + call_stream = LiveRunnerCallStream( status=resp.status, headers=resp.headers, runner_url=runner_url, @@ -835,6 +853,19 @@ async def call_runner( _session=session, _response=resp, ) + # Metered pricing: the stream outlives this call, so it owns + # the payment loop; aclose() stops both. + if payment_session is not None and payment_type in _METERED_PAYMENT_TYPES: + call_stream._payment_task = asyncio.create_task( + _run_call_payments( + payment_session, + interval_s=_payment_cadence_s( + challenge.payment_interval_s if challenge is not None else None + ), + on_released=call_stream._mark_released, + ) + ) + return call_stream # Metered pricing: the orchestrator debits the prepaid balance # while the request runs, so keep funding it for as long as we From f7ddca04b4ab3f04f4f7f6faa70e20bb64df737d Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 13:40:41 +0200 Subject: [PATCH 4/5] feat(live-runner): pay metered single-shot calls via the session-scoped endpoint Derive .../apps/{runner_id}/session/{session_id}/payment from the single-shot runner URL and the challenge manifest_id so the loop gets liveness semantics (404 once the session is released, 409 for fixed pricing) instead of blind-crediting the generic /payment endpoint, which stays as the fallback when the URL shape does not match. Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/live_runner.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index d9e312d..dd88536 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -862,6 +862,7 @@ async def call_runner( interval_s=_payment_cadence_s( challenge.payment_interval_s if challenge is not None else None ), + payment_url=_single_shot_payment_url(runner_url, session_id), on_released=call_stream._mark_released, ) ) @@ -878,6 +879,7 @@ async def call_runner( interval_s=_payment_cadence_s( challenge.payment_interval_s if challenge is not None else None ), + payment_url=_single_shot_payment_url(runner_url, session_id), ) ) try: @@ -961,6 +963,29 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC ) +def _single_shot_payment_url(runner_url: str, session_id: str) -> str: + """Derive the session-scoped payment endpoint for a single-shot call. + + Single-shot runner URLs are orchestrator proxy routes of the form + ``…/apps/{runner_id}/app[/…]``; the matching payment endpoint is + ``…/apps/{runner_id}/session/{session_id}/payment``, which 404s once the + session is released and 409s for fixed pricing. Returns "" when the URL + does not match that shape, in which case payments fall back to the + orchestrator's generic ``/payment`` endpoint. + """ + if not session_id: + return "" + try: + parsed = urlparse(_normalize_http_base(runner_url)) + except LivepeerGatewayError: + return "" + match = re.match(r"^(?P.*/apps/[^/]+)/app(?:/.*)?$", parsed.path) + if match is None: + return "" + path = f"{match.group('base')}/session/{quote(session_id, safe='')}/payment" + return urlunparse((parsed.scheme, parsed.netloc, path, "", "", "")) + + def _payment_cadence_s(server_interval_s: Optional[float]) -> float: if server_interval_s is not None and server_interval_s > 0: return server_interval_s * _PAYMENT_CADENCE_FRACTION From b0f46d4230778d243a0a8301df91333c6b25c96e Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 13:44:41 +0200 Subject: [PATCH 5/5] docs(examples): SDK client for the text demo with self-funding payments Streams the story via call_runner(stream=True); with --signer-url the call pays its 402 challenge and keeps funding itself while the stream runs, demonstrating that metered single-shot calls need no payment code on the caller side. Co-Authored-By: Claude Fable 5 --- examples/text/README.md | 9 ++++++++ examples/text/client.py | 48 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+) create mode 100644 examples/text/client.py diff --git a/examples/text/README.md b/examples/text/README.md index 7720e9f..f48eff0 100644 --- a/examples/text/README.md +++ b/examples/text/README.md @@ -37,3 +37,12 @@ Verify the runner registration with go-livepeer: ``` curl http://localhost:8935/discovery | jq ``` + +Or stream through the SDK client, which self-funds metered calls when a +remote signer is given (pays the 402 challenge, then keeps paying on a +cadence for as long as the stream runs): + +```sh +uv run client.py http://localhost:8935/apps/story-runner/app/sse +uv run client.py https://orch:8935/apps/story-runner/app/sse --signer-url http://localhost:7936 +``` diff --git a/examples/text/client.py b/examples/text/client.py new file mode 100644 index 0000000..8e65820 --- /dev/null +++ b/examples/text/client.py @@ -0,0 +1,48 @@ +"""SDK client for the text stream demo: a self-funding single-shot SSE call. + +Offchain (no payments): + + uv run client.py http://localhost:8935/apps/story-runner/app/sse + +On-chain: pass a remote signer. call_runner pays the 402 challenge and, +for metered pricing (hour/seconds or 720p), keeps funding the call on a +cadence while the stream runs — no payment code needed here. + + uv run client.py https://orch:8935/apps/story-runner/app/sse \ + --signer-url http://localhost:7936 +""" +from __future__ import annotations + +import argparse +import asyncio + +from livepeer_gateway.live_runner import call_runner + + +async def main() -> None: + p = argparse.ArgumentParser(description="Stream a story from a single-shot runner.") + p.add_argument("runner_url", help="Runner app endpoint, e.g. .../apps/story-runner/app/sse") + p.add_argument("--signer-url", default=None, help="Remote signer URL. Omit for offchain.") + p.add_argument( + "--payment-unit", + default=None, + help="hour|seconds|720p|720p-pixel-seconds|fixed. Default: metered (live).", + ) + args = p.parse_args() + + stream = await call_runner( + args.runner_url, + method="GET", + signer_url=args.signer_url, + payment_unit=args.payment_unit, + stream=True, + ) + async with stream: + async for line in stream.aiter_lines(): + print(line) + if stream.released: + print("(session released by orchestrator)") + + +if __name__ == "__main__": + asyncio.run(main())