YoutubeLiveCaptionSender

id: lib/sender

Direct caption delivery to YouTube Live via Google’s HTTP POST ingestion API.

Import

import { YoutubeLiveCaptionSender } from 'lcyt';
// CJS
const { YoutubeLiveCaptionSender } = require('lcyt');

Constructor

new YoutubeLiveCaptionSender(options)

Methods

start()

Initialize the sender. Call this before sending any captions.

await sender.start();

Returns: Promise<void>

send(text, timestamp?)

Send a single caption to YouTube.

const result = await sender.send('Hello, world!');
const result = await sender.send('Hello!', '2024-01-01T12:00:00.000');

Returns: Promise<SendResult>

interface SendResult {
  sequence: number;        // Sequence number used for this request
  timestamp: string;       // ISO timestamp sent to YouTube
  statusCode: number;      // HTTP status code from YouTube
  response: string;        // Raw response body from YouTube
  serverTimestamp: string; // Server-side timestamp (from YouTube response headers)
}

Throws: NetworkError on non-2xx response.

sendBatch(captions)

Send multiple captions in a single HTTP request.

const result = await sender.sendBatch([
  { text: 'First line', timestamp: '2024-01-01T12:00:00.000' },
  { text: 'Second line', timestamp: '2024-01-01T12:00:02.000' },
]);

Returns: Promise<BatchSendResult>

interface BatchSendResult {
  sequence: number;        // Sequence number of the last caption in the batch
  count: number;           // Number of captions sent
  statusCode: number;      // HTTP status code from YouTube
  response: string;        // Raw response body
  serverTimestamp: string; // Server-side timestamp
}

Throws: NetworkError on non-2xx response.

construct(text, timestamp?)

Add a caption to the internal queue without sending it.

sender.construct('Caption 1');
sender.construct('Caption 2', '2024-01-01T12:00:05.000');

Returns: void

getQueue()

Return the current internal caption queue.

const queue = sender.getQueue();
// [{ text: 'Caption 1', timestamp: '...' }, ...]

Returns: Array<{text: string, timestamp: string}>

clearQueue()

Clear all captions from the internal queue.

sender.clearQueue();

Returns: void

heartbeat()

Send an empty caption request to test connectivity without advancing captions.

const result = await sender.heartbeat();

Returns: Promise<HeartbeatResult>

interface HeartbeatResult {
  sequence: number;
  statusCode: number;
  serverTimestamp: string;
}

Throws: NetworkError on failure.

sync()

Perform an NTP-style clock synchronisation against the YouTube server.

Updates the internal syncOffset used to compensate for local/server clock drift.

const result = await sender.sync();
console.log(result.syncOffset); // e.g. 150 (ms)

Returns: Promise<SyncResult>

interface SyncResult {
  syncOffset: number;       // Computed offset in milliseconds (positive = server is ahead)
  roundTripTime: number;    // Round-trip latency in milliseconds
  serverTimestamp: string;  // ISO timestamp returned by YouTube
  statusCode: number;
}

end()

Flush pending captions and clean up resources.

await sender.end();

Returns: Promise<void>

getSequence() / setSequence(seq)

Read or write the internal sequence counter.

const seq = sender.getSequence(); // number
sender.setSequence(42);

getSyncOffset() / setSyncOffset(offset)

Read or write the clock synchronisation offset (milliseconds).

const offset = sender.getSyncOffset(); // number
sender.setSyncOffset(200);

Example: Full Workflow

import { YoutubeLiveCaptionSender } from 'lcyt';

const sender = new YoutubeLiveCaptionSender({
  streamKey: process.env.STREAM_KEY,
  verbose: true,
});

await sender.start();

// Synchronise clock
await sender.sync();

// Send individual captions
await sender.send('Welcome to the stream!');
await sender.send('Captions powered by lcyt.');

// Build a batch from a queue
sender.construct('Line one');
sender.construct('Line two');
await sender.sendBatch(sender.getQueue());
sender.clearQueue();

await sender.end();

BackendCaptionSender

id: lib/backend-sender

Relay-based caption sender that routes captions through an lcyt-backend HTTP server instead of calling YouTube directly. Mirrors the YoutubeLiveCaptionSender API.

Import

import { BackendCaptionSender } from 'lcyt/backend';
// CJS
const { BackendCaptionSender } = require('lcyt/backend');

Why Use the Relay?

• Hides your YouTube stream key behind an authenticated server
• Enables usage tracking, daily/lifetime limits, and GDPR controls
• Provides real-time caption delivery results via SSE (GET /events)
• Supports browser-based clients that cannot safely store a stream key

Constructor

new BackendCaptionSender(options)

Methods

start()

Register a session with the backend. Exchanges your API key + stream key for a JWT that is used for all subsequent requests.

await sender.start();

Returns: Promise<void>

Throws: NetworkError if the backend rejects registration.

send(text, timestamp?, extraOpts?)

Queue a single caption for delivery. Returns immediately with 202 Accepted; the actual YouTube delivery result arrives on the SSE event stream (GET /events).

const result = await sender.send('Hello!');
// With translation metadata:
const result = await sender.send('Hello!', undefined, {
  translations: { 'fi-FI': 'Hei!' },
  captionLang: 'fi-FI',
  showOriginal: true,
});

Timestamp formats

• ISO string: '2024-01-01T12:00:00.000'
• Unix milliseconds: 1704067200000
• Relative: { time: 5000 } — 5 seconds after the session startedAt (resolved by the server)

When translations are provided, the backend composes the final caption text and (if enabled on the API key) writes the original and each translation to separate files. See the backend /captions docs for the full composition rules.

Returns: Promise<{ok: true, requestId: string}>

The requestId correlates to the caption_result / caption_error SSE event.

sendBatch(captions)

Send multiple captions in one request.

const result = await sender.sendBatch([
  { text: 'Line one' },
  { text: 'Line two', timestamp: { time: 3000 } },
]);

Returns: Promise<{ok: true, requestId: string}>

construct(text, timestamp?)

Add a caption to the internal queue without sending.

sender.construct('Buffered caption');

getQueue() / clearQueue()

Read or empty the internal caption queue.

const queue = sender.getQueue();
sender.clearQueue();

heartbeat()

Test connectivity to the backend without sending a real caption.

await sender.heartbeat();

Returns: Promise<{ok: true}>

sync()

Perform a clock synchronisation round-trip via POST /sync on the backend.

const result = await sender.sync();

Returns: Promise<{syncOffset: number, roundTripTime: number, serverTimestamp: string, statusCode: number}>

updateSession(fields)

Update session metadata on the backend via PATCH /live.

await sender.updateSession({ sequence: 10 });

Returns: Promise<{sequence: number}>

getStartedAt()

Return the session start timestamp (set by the backend when the session was created).

const startedAt = sender.getStartedAt(); // ISO string

Returns: string | undefined

getSequence() / setSequence(seq)

Read or set the local sequence counter.

getSyncOffset() / setSyncOffset(offset)

Read or set the clock synchronisation offset (milliseconds).

end()

Tear down the backend session (DELETE /live).

await sender.end();

Returns: Promise<void>

SSE Event Stream

After calling start(), connect to the backend’s event stream to receive asynchronous delivery results:

const url = `${backendUrl}/events?token=${jwtToken}`;
const es = new EventSource(url);

es.addEventListener('caption_result', (e) => {
  const { requestId, sequence, statusCode } = JSON.parse(e.data);
  console.log('Delivered:', requestId, 'seq', sequence, 'status', statusCode);
});

es.addEventListener('caption_error', (e) => {
  const { requestId, error, statusCode } = JSON.parse(e.data);
  console.error('Failed:', requestId, error);
});

See the Backend API SSE docs for the full event reference.

Example

import { BackendCaptionSender } from 'lcyt/backend';

const sender = new BackendCaptionSender({
  backendUrl: 'https://relay.example.com',
  apiKey: 'my-api-key',
  streamKey: 'xxxx-xxxx-xxxx-xxxx',
  domain: 'https://my-app.example.com',
});

await sender.start();
await sender.sync();

const { requestId } = await sender.send('Hello from the relay!');
console.log('Queued with requestId:', requestId);

await sender.end();

Configuration

id: lib/config

Utilities for loading, saving, and building YouTube ingestion URLs from the lcyt configuration file (~/.lcyt-config.json).

Import

import { loadConfig, saveConfig, buildIngestionUrl, getDefaultConfigPath, getDefaultConfig } from 'lcyt/config';
// CJS
const { loadConfig, saveConfig, buildIngestionUrl } = require('lcyt/config');

Config File

By default, configuration is stored at ~/.lcyt-config.json. The file is plain JSON with the following shape:

{
  "baseUrl": "http://upload.youtube.com",
  "streamKey": "",
  "region": "us",
  "cue": "",
  "sequence": 0
}

Functions

getDefaultConfigPath()

Return the default path to the configuration file.

const path = getDefaultConfigPath();
// '/home/alice/.lcyt-config.json'

Returns: string

getDefaultConfig()

Return a configuration object populated with default values.

const config = getDefaultConfig();
// { baseUrl: 'http://upload.youtube.com', streamKey: '', region: 'us', cue: '', sequence: 0 }

Returns: LCYTConfig

loadConfig(path?)

Load configuration from a JSON file. Falls back to defaults for any missing field.

const config = loadConfig();                          // default path
const config = loadConfig('/custom/path/config.json'); // custom path

Returns: LCYTConfig

Throws: ConfigError if the file exists but cannot be parsed.

saveConfig(path?, config)

Persist a configuration object to disk as JSON.

saveConfig({ ...config, streamKey: 'xxxx-xxxx-xxxx-xxxx' });
saveConfig('/custom/path/config.json', { ...config, region: 'eu' });

Returns: void

Throws: ConfigError if the file cannot be written.

buildIngestionUrl(config)

Construct the full YouTube caption ingestion URL from a configuration object.

const url = buildIngestionUrl({
  baseUrl: 'http://upload.youtube.com',
  streamKey: 'xxxx-xxxx-xxxx-xxxx',
  region: 'us',
  cue: '',
  sequence: 0,
});
// 'http://upload.youtube.com/closedcaption?cid=xxxx-xxxx-xxxx-xxxx&region=us&...'

Returns: string — Full ingestion URL

TypeScript Type

interface LCYTConfig {
  baseUrl: string;
  streamKey: string;
  region: string;
  cue: string;
  sequence: number;
}

Example: CLI-style Config Merge

import { loadConfig, saveConfig, buildIngestionUrl } from 'lcyt/config';

// Load existing config
const config = loadConfig();

// Override with CLI arguments
if (process.argv[2]) config.streamKey = process.argv[2];

// Persist updated config
saveConfig(config);

// Build the ingestion URL
const url = buildIngestionUrl(config);
console.log('Sending to:', url);

Logger

id: lib/logger

A pluggable, structured logger used throughout lcyt. All log output is prefixed with [LCYT].

Import

import logger from 'lcyt/logger';
// CJS
const logger = require('lcyt/logger').default;

The module exports a global singleton instance — all modules in the same process share the same logger state.

Log Methods

All methods accept a message string and any number of additional arguments (passed to the underlying output function).

logger.info('Starting sender...');
logger.success('Caption sent (seq 42)');
logger.warn('Stream key not set');
logger.error('Connection failed', err);
logger.debug('Raw response:', response);

Configuration Methods

setVerbose(enabled)

Enable or disable debug-level logging.

logger.setVerbose(true);
logger.debug('This will now appear');

setSilent(enabled)

Suppress all log output (useful for library consumers who handle output themselves).

logger.setSilent(true);

setUseStderr(enabled)

Route all log output to stderr instead of stdout.

logger.setUseStderr(true);

MCP servers must set this to true (or set LCYT_LOG_STDERR=1) because the MCP protocol uses stdout for its own messages. Writing logs to stdout will corrupt the MCP stream.

setCallback(fn)

Register a callback that receives every log event. Useful for piping logs into a UI or file.

logger.setCallback((level, message, ...args) => {
  myLogStore.push({ level, message, extra: args });
});

Environment Variable

Set this in the environment when running as an MCP server subprocess to avoid corrupting the stdio transport.

Example: Redirecting Logs to a File

import logger from 'lcyt/logger';
import fs from 'node:fs';

const logFile = fs.createWriteStream('./lcyt.log', { flags: 'a' });

logger.setCallback((level, message) => {
  logFile.write(`[${new Date().toISOString()}] [${level}] ${message}\n`);
});

logger.setSilent(true); // suppress stdout output

Example: Verbose Mode for Development

import logger from 'lcyt/logger';

logger.setVerbose(true);
logger.debug('This shows detailed internals'); // now visible

Error Classes

id: lib/errors

lcyt uses a typed error hierarchy so callers can handle errors at different levels of specificity. All errors extend the base LCYTError class.

Import

import { LCYTError, ConfigError, NetworkError, ValidationError } from 'lcyt/errors';
// CJS
const { LCYTError, ConfigError, NetworkError, ValidationError } = require('lcyt/errors');

Hierarchy

Error
└── LCYTError
    ├── ConfigError
    ├── NetworkError  (+ statusCode)
    └── ValidationError  (+ field)

LCYTError

Base class for all lcyt errors. Catch this to handle any library error.

try {
  await sender.send('text');
} catch (err) {
  if (err instanceof LCYTError) {
    console.error('lcyt error:', err.message);
  }
}

ConfigError

Thrown when a configuration file cannot be read, parsed, or written.

import { ConfigError } from 'lcyt/errors';
import { loadConfig } from 'lcyt/config';

try {
  const config = loadConfig('/bad/path.json');
} catch (err) {
  if (err instanceof ConfigError) {
    console.error('Config problem:', err.message);
  }
}

NetworkError

Thrown when an HTTP request to YouTube (or the relay backend) fails, either due to a transport error or a non-2xx status code.

import { NetworkError } from 'lcyt/errors';

try {
  await sender.send('Hello!');
} catch (err) {
  if (err instanceof NetworkError) {
    console.error(`HTTP ${err.statusCode}: ${err.message}`);
  }
}

ValidationError

Thrown when input values fail validation before a request is made.

import { ValidationError } from 'lcyt/errors';

try {
  await sender.send(''); // empty caption
} catch (err) {
  if (err instanceof ValidationError) {
    console.error(`Invalid field "${err.field}": ${err.message}`);
  }
}

Catching All Errors

import { LCYTError, NetworkError, ValidationError, ConfigError } from 'lcyt/errors';

try {
  await sender.send(text);
} catch (err) {
  if (err instanceof ValidationError) {
    // Input problem — fix the request
    console.error(`Bad input for field "${err.field}"`);
  } else if (err instanceof NetworkError) {
    // HTTP/transport problem — may be transient
    console.error(`Network error (${err.statusCode ?? 'no status'}):`, err.message);
  } else if (err instanceof ConfigError) {
    // Config problem — check ~/.lcyt-config.json
    console.error('Configuration error:', err.message);
  } else if (err instanceof LCYTError) {
    // Unknown lcyt error
    console.error('lcyt error:', err.message);
  } else {
    throw err; // unexpected error — rethrow
  }
}

YoutubeLiveCaptionSender (Python)

Direct caption delivery to YouTube Live via Google’s HTTP POST ingestion API.

Import

from lcyt.sender import YoutubeLiveCaptionSender, Caption, SendResult

Dataclasses

Caption

Represents a single caption for batch sending.

@dataclass
class Caption:
    text: str
    timestamp: str | datetime | int | float | None = None

SendResult

Returned by send(), send_batch(), and heartbeat().

@dataclass
class SendResult:
    sequence: int
    status_code: int
    response: str
    server_timestamp: str | None = None
    timestamp: str | None = None   # set by send() only
    count: int | None = None       # set by send_batch() only

Constructor

YoutubeLiveCaptionSender(
    stream_key=None,
    base_url=DEFAULT_BASE_URL,
    ingestion_url=None,
    region="reg1",
    cue="cue1",
    use_region=False,
    sequence=0,
    use_sync_offset=False,
    verbose=False,
)

Lifecycle Methods

start()

Initialise the sender. Must be called before sending captions.

sender.start()
# or chained:
sender = YoutubeLiveCaptionSender(stream_key="...").start()

Returns: self (for method chaining)

Raises: ValidationError if neither stream_key nor ingestion_url is set.

end()

Stop the sender and clear the internal queue.

sender.end()

Returns: self

Sending Methods

send(text, timestamp=None)

Send a single caption to YouTube.

result = sender.send("Hello, world!")
result = sender.send("Hello!", "2024-01-01T12:00:00.000")
result = sender.send("Recent", -2.0)  # 2 seconds ago (relative)

Returns: SendResult

Raises: ValidationError if sender not started or text is empty. NetworkError on HTTP failure.

send_batch(captions=None)

Send a list of captions in one HTTP request.

from lcyt.sender import Caption

result = sender.send_batch([
    Caption(text="First line", timestamp="2024-01-01T12:00:00.000"),
    Caption(text="Second line", timestamp="2024-01-01T12:00:02.000"),
])

If captions is None, the internal queue (built with construct()) is drained and sent.

Returns: SendResult with count set to the number of captions sent.

Raises: ValidationError if no captions to send. NetworkError on HTTP failure.

construct(text, timestamp=None)

Add a caption to the internal queue without sending it.

sender.construct("Caption 1")
sender.construct("Caption 2", "2024-01-01T12:00:05.000")
# then send the queue:
sender.send_batch()

Returns: int — current queue length.

Raises: ValidationError if text is empty or not a string.

heartbeat()

Send an empty POST to verify connectivity without advancing captions.

Per Google’s spec the heartbeat does not increment the sequence number.

result = sender.heartbeat()
print(result.status_code)  # 200

Returns: SendResult

Raises: NetworkError on failure.

sync()

Perform an NTP-style clock synchronisation against the YouTube server.

Sends a heartbeat, measures round-trip time, and computes the clock offset between the local clock and YouTube’s server timestamp. Automatically sets use_sync_offset = True so future auto-generated timestamps are corrected.

info = sender.sync()
print(info["sync_offset"])      # e.g. 150 (ms)
print(info["round_trip_time"])  # e.g. 82 (ms)

Returns: dict with keys:

Raises: NetworkError on failure.

send_test()

Send a test payload using Google’s region:reg1#cue1 format.

result = sender.send_test()
print(result.status_code)  # 200 if connection is working

Returns: SendResult

Queue Management

get_queue()

Return a copy of the internal caption queue.

queue = sender.get_queue()
# [Caption(text='Caption 1', timestamp=None), ...]

Returns: list[Caption]

clear_queue()

Clear all captions from the internal queue.

count = sender.clear_queue()
# int — number of captions cleared

Returns: int

Sequence Management

get_sequence() / set_sequence(sequence)

Read or write the internal sequence counter.

seq = sender.get_sequence()   # int
sender.set_sequence(42)       # returns self

Sync Offset Management

get_sync_offset() / set_sync_offset(offset)

Read or write the clock synchronisation offset (milliseconds).

offset = sender.get_sync_offset()   # int
sender.set_sync_offset(200)         # returns self

Properties

Example: Full Workflow

from lcyt.sender import YoutubeLiveCaptionSender, Caption
import os

sender = YoutubeLiveCaptionSender(
    stream_key=os.environ["STREAM_KEY"],
    verbose=True,
)

sender.start()

# Synchronise clock
info = sender.sync()
print(f"Sync offset: {info['sync_offset']}ms")

# Send individual captions
sender.send("Welcome to the stream!")
sender.send("Captions powered by lcyt.")

# Build a batch from a queue
sender.construct("Line one")
sender.construct("Line two")
sender.send_batch()  # drains the queue

sender.end()

BackendCaptionSender (Python)

Send live captions via an lcyt-backend relay server instead of directly to YouTube.

Import

from lcyt.backend_sender import BackendCaptionSender

Overview

BackendCaptionSender communicates with an lcyt-backend HTTP server rather than YouTube’s ingestion endpoint directly. It mirrors the YoutubeLiveCaptionSender API but returns response dicts instead of SendResult dataclasses.

Why use this?

• Your client cannot reach YouTube directly (firewall, CORS, restricted network)
• You want multi-user session management and API key enforcement from the relay
• You need the SSE result stream (GET /events) for async delivery confirmation

Async delivery: send() returns immediately with {"ok": True, "requestId": "..."}. The actual YouTube delivery result arrives on the GET /events SSE stream.

Constructor

BackendCaptionSender(
    backend_url,
    api_key,
    stream_key,
    domain="http://localhost",
    sequence=0,
    verbose=False,
)

Lifecycle Methods

start()

Register a session with the backend and obtain a JWT token.

sender.start()
# or chained:
sender = BackendCaptionSender(...).start()

Updates internal sequence, sync_offset, and started_at from the server response. Idempotent — returns the existing session if one already exists.

Returns: self

Raises: NetworkError on HTTP failure.

end()

Tear down the backend session and clear the stored JWT.

sender.end()

Returns: self

Raises: NetworkError on HTTP failure.

Sending Methods

send(text, timestamp=None, time=None)

Send a single caption via the backend.

result = sender.send("Hello, world!")
result = sender.send("Absolute time", timestamp="2024-01-01T12:00:00.000")
result = sender.send("Relative time", time=5000)  # 5 sec since session start

Returns: dict — {"ok": True, "requestId": "..."} (202 Accepted)

Raises: NetworkError on HTTP failure.

send_batch(captions=None)

Send multiple captions in one request.

result = sender.send_batch([
    {"text": "Line one"},
    {"text": "Line two", "timestamp": "2024-01-01T12:00:02.000"},
    {"text": "Line three", "time": 5000},
])

If captions is None, drains and sends the internal queue (built with construct()).

Returns: dict — {"ok": True, "requestId": "..."}

Raises: NetworkError on HTTP failure.

construct(text, timestamp=None, time=None)

Add a caption to the local queue without sending.

sender.construct("Caption 1")
sender.construct("Caption 2", time=3000)
sender.send_batch()  # flush the queue

Returns: int — current queue length.

Queue Management

get_queue()

Return a copy of the local queue.

queue = sender.get_queue()
# [{"text": "Caption 1"}, {"text": "Caption 2", "time": 3000}]

Returns: list[dict]

clear_queue()

Clear the local queue.

count = sender.clear_queue()  # int — items cleared

Returns: int

Sync and Heartbeat

sync()

Trigger an NTP-style clock sync on the backend. Updates the local sync_offset.

data = sender.sync()
print(data["syncOffset"])      # ms offset
print(data["roundTripTime"])   # ms

Returns: dict — {"syncOffset": int, "roundTripTime": int, "serverTimestamp": str, "statusCode": int}

Raises: NetworkError on failure.

heartbeat()

Check the session status on the backend. Updates local sequence and sync_offset.

data = sender.heartbeat()
print(data["sequence"])

Returns: dict — {"sequence": int, "syncOffset": int}

Raises: NetworkError on failure.

Getters / Setters

Example: Full Workflow

from lcyt.backend_sender import BackendCaptionSender
import os

sender = BackendCaptionSender(
    backend_url=os.environ["BACKEND_URL"],
    api_key=os.environ["API_KEY"],
    stream_key=os.environ["STREAM_KEY"],
    domain="https://my-app.example.com",
)

sender.start()
sender.sync()

sender.send("Welcome to the stream!")

# Queue and batch-send
sender.construct("Line one")
sender.construct("Line two", time=3000)
sender.send_batch()

sender.end()

Configuration (Python)

Utilities for loading, saving, and building YouTube ingestion URLs from the lcyt configuration file (~/.lcyt-config.json).

Import

from lcyt.config import (
    LCYTConfig,
    load_config,
    save_config,
    build_ingestion_url,
    get_default_config_path,
)

Config File

By default, configuration is stored at ~/.lcyt-config.json. The file is plain JSON:

{
  "stream_key": "",
  "base_url": "http://upload.youtube.com/closedcaption",
  "region": "reg1",
  "cue": "cue1",
  "sequence": 0
}

The Python library accepts both snake_case and camelCase keys when reading — making the config file interoperable with the Node.js library.

LCYTConfig Dataclass

@dataclass
class LCYTConfig:
    stream_key: str = ""
    base_url: str = "http://upload.youtube.com/closedcaption"
    region: str = "reg1"
    cue: str = "cue1"
    sequence: int = 0

Methods

Functions

get_default_config_path()

Return the default config file path.

path = get_default_config_path()
# PosixPath('/home/alice/.lcyt-config.json')

Returns: Path

load_config(config_path=None)

Load configuration from a JSON file. Returns defaults for any missing field.

config = load_config()                                # default path
config = load_config("/custom/path/config.json")     # custom path
config = load_config(Path("/custom/path/config.json"))

Returns: LCYTConfig

Raises: ConfigError if the file exists but cannot be read or parsed.

save_config(config, config_path=None)

Persist a LCYTConfig instance to disk as JSON.

save_config(config)                              # default path
save_config(config, "/custom/path/config.json") # custom path

Returns: None

Raises: ConfigError if the file cannot be written.

build_ingestion_url(config)

Construct the full YouTube caption ingestion URL from a config object.

url = build_ingestion_url(config)
# 'http://upload.youtube.com/closedcaption?cid=xxxx-xxxx-xxxx-xxxx'

Returns: str — full ingestion URL

Raises: ConfigError if stream_key is empty.

Example: CLI-style Config Merge

import sys
from lcyt.config import load_config, save_config, build_ingestion_url

# Load existing config
config = load_config()

# Override with CLI argument
if len(sys.argv) > 1:
    config.stream_key = sys.argv[1]

# Persist updated config
save_config(config)

# Build the ingestion URL
url = build_ingestion_url(config)
print("Sending to:", url)

Error Classes (Python)

lcyt uses a typed exception hierarchy so callers can handle errors at different levels of specificity. All exceptions extend the base LCYTError class.

Import

from lcyt.errors import LCYTError, ConfigError, NetworkError, ValidationError

Hierarchy

Exception
└── LCYTError
    ├── ConfigError
    ├── NetworkError  (+ status_code)
    └── ValidationError  (+ field)

LCYTError

Base class for all lcyt exceptions. Catch this to handle any library error.

from lcyt.errors import LCYTError

try:
    result = sender.send("text")
except LCYTError as e:
    print("lcyt error:", e)

ConfigError

Raised when a configuration file cannot be read, parsed, or written.

from lcyt.errors import ConfigError
from lcyt.config import load_config

try:
    config = load_config("/bad/path.json")
except ConfigError as e:
    print("Config problem:", e)

NetworkError

Raised when an HTTP request to YouTube (or the relay backend) fails, either due to a transport error or a non-2xx status code.

from lcyt.errors import NetworkError

try:
    result = sender.send("Hello!")
except NetworkError as e:
    print(f"HTTP {e.status_code}: {e}")

ValidationError

Raised when input values fail validation before a request is made.

from lcyt.errors import ValidationError

try:
    sender.send("")  # empty text
except ValidationError as e:
    print(f"Invalid field '{e.field}': {e}")

Catching All Errors

from lcyt.errors import LCYTError, NetworkError, ValidationError, ConfigError

try:
    result = sender.send(text)
except ValidationError as e:
    # Input problem — fix the request
    print(f"Bad input for field '{e.field}'")
except NetworkError as e:
    # HTTP/transport problem — may be transient
    print(f"Network error ({e.status_code}): {e}")
except ConfigError as e:
    # Config problem — check ~/.lcyt-config.json
    print("Configuration error:", e)
except LCYTError as e:
    # Unknown lcyt error
    print("lcyt error:", e)