License

The lcyt core library (Node.js and Python) and the lcyt-cli command-line tool are licensed under the European Union Public Licence (EUPL) v. 1.2. This page reproduces the licence text in full; the canonical copy lives in each package’s LICENSE file (packages/lcyt/LICENSE, packages/lcyt-cli/LICENSE) and at eupl.eu/1.2/en.

Other parts of the LCYT monorepo (backends, web UI, plugins, etc.) may be licensed differently — check each package’s own LICENSE file.

European Union Public Licence (EUPL) v. 1.2

Terms and Conditions

This European Union Public Licence (the “Licence”) applies to the Work (as defined below) which is provided under the terms of this Licence. Any use of the Work, other than as authorised under this Licence is prohibited (to the extent such use is covered by a right of the copyright holder of the Work).

The Work is provided under the terms of this Licence when the Licensor (as defined below) has placed the following notice immediately following the copyright notice for the Work:

“Licensed under the EUPL”

or has expressed it in any other manner.

1. Definitions

In this Licence, the following terms have the following meaning:

• The “Licence”: this licence.
• The “Original Work”: the work or software distributed or communicated by the Licensor under this Licence, including source code and documentation, and any updates or upgrades thereof unless they constitute Derivative Works. If the Original Work is a database, then any updates or upgrades thereof exclude the contents of the database for the purposes of defining the Original Work. If the Original Work is a source code repository, then any updates or upgrades consisting of data contained in the repository itself are excluded from the definition of the Original Work; for the purposes of this definition, a repository is a location where data is stored and managed, whether or not it is accessible via electronic networks.
• The “Derivative Works”: any work or software, other than the Original Work, that is based on (or derived from) the Original Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship, of which the copyright is held by the Licensor or, where applicable, any third party. This Licence does not define the degree of modification or dependence on the Original Work required to classify a work as a Derivative Work; this degree is determined by copyright law of the country mentioned in Article 15.
• The “Work”: the Original Work or its Derivative Works.
• The “Licensor”: the natural or legal person distributing or communicating the Work under this Licence.
• “Distribution” or “Communication”: any act of selling, giving, lending, renting, transmitting, or otherwise making available, online or offline, copies of the Work or providing access to its essential functionalities at a distance by any person to any other person. For the purposes of this definition, transmission of the Work as part of a larger work in which the Work is incorporated shall not constitute Distribution or Communication.
• “You” or “Licensee”: the natural or legal person exercising rights under the Licence.
• “Source code”: the human-readable form of a work that is the most convenient for people to study and modify it.
• “Object code”: any form of a work that results from the transformation or translation of source code, including but not limited to compiled object code and intermediate forms such as bytecode.

2. Scope of the Licence

The rights granted to You under this Licence are subject to the following scope: The Licensor hereby grants You a worldwide, royalty-free, non-exclusive, sublicensable licence to do the following, for the duration of copyright vested in the Original Work:

(a) use the Work in any circumstance and for all usage, (b) reproduce the Work, (c) modify the Work, and develop Derivative Works based upon the Work, (d) communicate to the public, including the right to make available or display the Work or copies thereof to the public and perform publicly, as the case may be, the Work, (e) distribute the Work or copies thereof, (f) lend and rent the Work or copies thereof, (g) sublicense rights in the Work or copies thereof.

These rights can be exercised on any media, supports and formats, whether now known or later invented, as far as the applicable law permits so.

In the countries where moral rights apply, the Licensor waives the right to exercise the moral right of paternity of the Work to the fullest extent allowed by law so as to allow for the effective exercise of the economic rights aforesaid.

The Licensor grants to the Licensee royalty-free, non-exclusive usage rights to any patents held by the Licensor, to the extent necessary to make use of the rights granted on the Work under this Licence.

3. Communication of Source Code

The Licensor may provide the Work either in its source code form, or as object code. If the Work is provided as object code, the Licensor provides in addition a machine-readable copy of the source code of the Work along with each copy of the Work that the Licensor distributes or indicates, in a notice following the copyright notice attached to the Work, a repository where the source code is easily and freely accessible for as long as the Licensor continues to distribute or communicate the Work.

4. Limitations on copyright

Nothing in this Licence is intended to deprive the Licensee of the benefits from any exception or limitation to the exclusive rights of the rights owners in the Original Work or Derivative Works, nor does it constitute a waiver of any of the said benefits and limitations.

5. Obligations of the Licensee

The grant of the rights mentioned above is subject to some restrictions and obligations imposed on the Licensee. Those obligations are the following:

Attribution right: The Licensee shall keep intact all copyright, patent or trademarks notices and all notices that refer to the Licence and to the disclaimer of warranties. The Licensee must include a copy of such notices and a copy of the Licence with every copy of the Work he/she distributes or communicates. The Licensee must cause any Derivative Work to carry prominent notices stating that the Work has been modified and the date of modification.

Copyleft clause: If the Licensee distributes or communicates copies of the Original Works or Derivative Works thereof, this Distribution or Communication will be done under the terms of this Licence or of a later version of this Licence unless the Original Work is expressly distributed only under this version of the Licence — for example by the communication of “EUPL v. 1.2 only”. The Licensee (becoming Licensor) cannot offer or impose any additional terms or conditions on the Work or Derivative Work that alter or restrict the terms of the Licence.

Compatibility clause: If the Licensee distributes or communicates Derivative Works or copies thereof based upon both the Work and another work licensed under a compatible licence, this Distribution or Communication can be done under the terms of this compatible licence. For the sake of this clause, “compatible licence” refers to the licences listed in the appendix attached to this Licence. Should the Licensee’s obligations under the compatible licence conflict with his/her obligations under this Licence, the obligations of the compatible licence shall prevail.

Provision of Source Code: When distributing or communicating copies of the Work, the Licensee will provide a machine-readable copy of the source code or indicate a repository where this source is easily and freely accessible for as long as the Licensee continues to distribute or communicate the Work.

Legal Protection of the Licensee: Any Licensee which finally and non-provisionally terminates its violations of this Licence is reinstated with all his rights under this Licence. The reinstatement of rights is permanent. However, if the violation continues after a year following the notice of the non-definitive termination and a cure, termination of the Licence is permanent.

6. Disclaimer of Warranties

The Work is a work in progress, which is continuously improved by numerous contributors. It is not a finished work and may therefore contain defects or bugs inherent to this type of development.

For the above reason, the Work is provided under the Licence on an “as-is” basis and without warranties of any kind concerning the Work, including without limitation merchantability, fitness for a particular purpose, absence of defects or errors, accuracy, non-infringement of intellectual property rights other than as stated in Article 4 of this Licence.

This disclaimer of warranties is an essential part of the Licence and a condition for the grant of any rights to the Work.

7. Limitation of Liability

Subject to the provisions of Article 8, the Licensor shall not be liable for any direct or indirect, material or moral, damages of any kind, arising out of the Licence or of the use of the Work, including without limitation, damages for loss of goodwill, work stoppage, computer failure or malfunction, loss of data or any commercial damage, even if the Licensor has been advised of the possibility of such damage. However, the Licensor will be liable under statutory product liability laws as far such laws apply to the Work.

8. Limitation of Liability

The limitation of liability shall not apply to:

(a) liability for death or personal injury resulting from negligence attributable to the Licensor, (b) liability for direct damages resulting from any breach of the essential obligations in this Licence or from violation of intellectual property rights by the Licensor, since these are the reasons for which the Licensor is party to this Licence.

For the sake of clarity, this Licence is a licence of copyright held by the Licensor and does not transfer any ownership of the Work.

9. Miscellaneous

(a) Each time You accept the Licence, the original Licensor offers You a licence to the Work under the same terms and conditions. (b) If any provision of this Licence is invalid or unenforceable under applicable law, this shall not affect the validity or enforceability of the Licence as a whole. Such provision shall be interpreted or reformed so as to achieve the goals and purposes of that provision under applicable law. (c) Without prejudice to any other proceedings which may be available to him/her, any Licensor revoking this Licence for your material breach of the terms and conditions of this Licence by a written notice addressed to you, will not be liable for damages resulting from loss of opportunity (loss of revenue, profit and anticipated savings) or for indirect damages, consequential damages or punitive damages as regards such revocation. (d) As the Work is continuously developed, assistance is often provided by various contributors (“Contributors”). Except in the case provided for in Article 5(c), the Contributors do not benefit from any exclusive rights and each Contributor retains copyright to modifications made by him/her.

10. Versions of the Licence

10.1. New Versions

The European Commission may publish new versions and/or new language versions of this Licence or updated versions of the Appendix, so far this is required to address new developments, further to the results of the Community legal framework relating to information society technology. Every version of the Licence will be given a distinguishing version number.

10.2. Effect of New Versions

Any Work already distributed or communicated by a Licensor under this version of the Licence may continue to be distributed or communicated thereunder, as long as the Derivative Works are distributed under the same version of the Licence.

10.3. Earlier Versions

You may choose to use the Work under the terms of either this Licence or any earlier version, unless the Licence explicitly excludes the use under the terms of the earlier versions. For any version of the Licence, you will find in the Appendix all the licenses with which that version is considered compatible.

11. Jurisdiction

Without prejudice to specific agreement between parties,

(a) any litigation resulting from the interpretation of this License, arising between the European Commission, as a Licensor, and any Licensee, will be subject to the jurisdiction of the Court of Justice of the European Communities, as laid down in article 292 of the Treaty establishing the European Community, (b) any litigation arising between other parties and resulting from the interpretation of this License, will be subject to the exclusive jurisdiction of the competent court where the Licensor resides.

12. Applicable Law

This Licence shall be governed by the law of the European Union country in which the Licensor resides.

However, this choice will not deprive the Licensee of the protection granted by mandatory provisions of the law of the country in which the Licensee resides.

13. Appendix

“Compatible Licences” according to Article 5 EUPL are:

• GNU General Public License (GPL) v. 2, v. 3
• GNU Affero General Public License (AGPL) v. 3
• Open Software License (OSL) v. 2.1, v. 3.0
• Eclipse Public License (EPL) v. 1.0
• CeCILL v. 2.0, v. 2.1
• Mozilla Public License (MPL) v. 2.0
• GNU Lesser General Public License (LGPL) v. 2.1, v. 3
• Creative Commons Attribution-ShareAlike 3.0 Unported

Copyright (c) 2026 Juha Itäleino

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)