Library API

The package is both a CLI and an importable Python library. Everything below is exported from the package root.

from cartesia_msteams_bridge import load_config, start_server

Run the bridge in your own service

load_config() reads the same environment variables as the CLI and raises a clear ValueError when a required variable is missing or a numeric one is not a number. start_server(cfg) is a coroutine that starts listening and returns a BridgeServer handle.

import asyncio
from cartesia_msteams_bridge import load_config, start_server

async def main():
    server = await start_server(load_config())
    print("bridge up")
    try:
        await asyncio.Event().wait()   # run until cancelled
    finally:
        await server.close()           # drains live calls (session.end + close)

asyncio.run(main())

server.drain() ends every live call gracefully without stopping the listener; server.close() drains and stops. A session mid-goodbye is left to finish (its hard-bounded backstop tears it down), so a rolling deploy never cuts off the last thing the caller hears. The CLI wires SIGTERM/SIGINT to this for you - in your own app, hook your shutdown path to server.close().

Custom agent transport (testing)

The connect_line argument to start_server is an async factory that returns an AgentPort. The default mints a per-call access token and opens a real Line stream; tests substitute a fake so no network or API key is needed.

from cartesia_msteams_bridge import load_config, start_server

async def fake_connector(cfg, log, handlers):
    class FakePort:
        is_open = True
        def send_start(self, start): ...
        def send_audio_chunk(self, b64): ...
        def send_dtmf(self, digit): ...
        def send_custom(self, metadata): ...
        def close(self): ...
    # push server->bridge events with handlers.on_message({"event": "ack"}) and
    # agent audio with handlers.on_audio(base64_pcm)
    return FakePort()

server = await start_server(load_config(), connect_line=fake_connector)

The repository’s own test suite uses exactly this shape - tests/conftest.py has a reusable FakeAgentPort.

Cartesia-side helpers

Exported for tooling and tests:

• CartesiaAgentSocket - the real Line socket (per-call token mint, start/ack, keepalive pings).
• build_start(stream_id, cfg, caller, call_id) - the start event builder (pcm_16000 pinned, overrides only when configured, caller context in metadata).
• mint_access_token(cfg) - POST /access-token with grants: {agent: true}.
• synthesize_goodbye(cfg, text) - standalone Sonic TTS returning raw 16 kHz PCM.
• WIRE_SAMPLE_RATE (16000) and INPUT_FORMAT (“pcm_16000”).

HMAC helpers

Useful if you build tools that talk to the bridge, or want to test the upgrade.

import time
from cartesia_msteams_bridge import sign, verify, is_fresh, TIMESTAMP_HEADER, SIGNATURE_HEADER

ts = int(time.time() * 1000)
signature = sign(secret, ts, call_id)   # HMAC-SHA256(secret, f"{ts}.{call_id}") hex
# send as headers X-StandIn-Timestamp / -Signature
verify(secret, ts, call_id, signature)  # constant-time, False on any missing/malformed input
is_fresh(ts, 60_000)                    # within the freshness window?

Protocol helpers

Wire messages are plain dicts (they arrive and leave as JSON). parse_worker_message(raw) is the guarded parser (returns None on junk), and pcm16k_bytes_to_ms(n) converts PCM byte counts to milliseconds. See the Wire Protocol for the full contract, including the media-timeline timestampMs semantics.

Also exported

• authorize_upgrade, call_id_from_path, ReplayGuard - the upgrade-authorization primitives.
• CallSession, WorkerPort, AgentPort, LineConnector, LineSessionHandlers - the per-call relay class and its transport protocols (advanced embedding).
• BridgeConfig, BridgeServer - the public types.
• load_dotenv - the tiny .env loader the CLI uses.
• render_metrics, reset_metrics, logger, Logger - metrics text and the minimal leveled logger.