# Livepeer Python SDK

The official Python SDK for the Livepeer network. Discover and call Live Runner
apps through orchestrators, use a remote signer for payment, and stream
media and control data over Livepeer's trickle protocol.

## Requirements

- Python 3.12 or newer

## Installation

Install the stable release from PyPI:

```bash
python -m pip install livepeer-gateway
```

The import package is named `livepeer_gateway`:

```python
from livepeer_gateway import call_runner, reserve_session, runner_selector
```

## Development

Install dependencies and run the tests with [uv](https://docs.astral.sh/uv/):

```bash
uv sync --locked --group test
uv run --group test pytest
```

## Usage Examples

Install the example dependencies:

```bash
uv sync --extra examples
```

- [Echo video runner](examples/echo/README.md): view input, loopback, and output,
  publish video, and update app parameters.
- [Ping/pong websocket runner](examples/ping-pong/README.md): discover a runner
  and exchange websocket messages.
- [Text streaming runner](examples/text/README.md): stream text and SSE through
  a statically configured runner.

Inspect a matching runner's discovery metadata:

```bash
uv run examples/select_runner.py http://localhost:8935 --app livepeer-sample/echo
```

Runner discovery uses explicit `orchestrators` (their `/discovery` endpoints),
then `discovery_url`, then the endpoint derived from `signer_url`.

Reserve a persistent session using a gateway token and release it on exit:

```python
import asyncio

from livepeer_gateway import parse_token, reserve_session


async def main(token):
    config = parse_token(token)
    async with await reserve_session(
        app="livepeer-sample/echo",
        orchestrators=config.get("orchestrators"),
        signer_url=config.get("signer"),
        signer_headers=config.get("signer_headers"),
        discovery_url=config.get("discovery"),
        discovery_headers=config.get("discovery_headers"),
    ) as session:
        print(session.app_url)


asyncio.run(main("<base64-token>"))
```

For single-shot apps, use `runner_selector()` and `await cursor.next()`, or
`call_runner()` for a known runner URL. Paid runners support `hour`, `seconds`,
and `fixed` pricing through a remote signer.

Read demuxed packets without decoding, using the output channel URL returned
by the app:

```python
from livepeer_gateway import MediaOutput


async def read_packets(output_url):
    async with MediaOutput(output_url) as output:
        async for packet in output.packets():
            print(packet.kind, packet.stream_index, packet.pts_time, packet.size)
```

## Token schema (base64-encoded JSON object)

| Field | Type | Description |
|---|---|---|
| `orchestrators` | `string[]` (optional) | Ordered, non-empty orchestrator addresses used instead of discovery |
| `signer` | `string` (optional) | Signer base URL |
| `signer_headers` | `{"key": "value"}` (optional) | Extra HTTP headers sent to all signer endpoints |
| `discovery` | `string` (optional) | Discovery endpoint URL |
| `discovery_headers` | `{"key": "value"}` (optional) | Extra HTTP headers sent to an explicit discovery endpoint |

Parse tokens with `parse_token()` and pass the resulting fields explicitly to
runner APIs, as shown above.