LiveKit — Telephony, Egress, Ingress & Production¶

# sip.yaml
api_key: APIexample
api_secret: <secret>
ws_url: ws://livekit:7880
redis:
  address: redis:6379          # the same Redis as the server
sip_port: 5060                 # signalling (UDP and TCP)
rtp_port: 10000-20000          # media (UDP)
use_external_ip: true
health_port: 8080
prometheus_port: 9090
log_level: info

docker run --rm --network host -e SIP_CONFIG_BODY="$(cat sip.yaml)" livekit/sip

import asyncio

from livekit import api


async def main() -> None:
    async with api.LiveKitAPI() as lk:                                   # LIVEKIT_URL / KEY / SECRET
        trunk = await lk.sip.create_sip_inbound_trunk(
            api.CreateSIPInboundTrunkRequest(
                trunk=api.SIPInboundTrunkInfo(
                    name="support-in",
                    numbers=["+15105550100"],
                    allowed_numbers=["+13105550100"],
                    auth_username="trunk-user",
                    auth_password="trunk-pass",
                )
            )
        )
        await lk.sip.create_sip_dispatch_rule(
            api.CreateSIPDispatchRuleRequest(
                name="support-calls",
                trunk_ids=[trunk.sip_trunk_id],
                rule=api.SIPDispatchRule(dispatch_rule_individual=api.SIPDispatchRuleIndividual(room_prefix="call-")),
                room_config=api.RoomConfiguration(
                    agents=[api.RoomAgentDispatch(agent_name="support", metadata='{"queue":"support"}')]
                ),
            )
        )


asyncio.run(main())

await lk.sip.create_sip_participant(
    api.CreateSIPParticipantRequest(
        sip_trunk_id="ST_xxx", sip_call_to="+15105550123",
        room_name="outbound-1", participant_identity="callee", participant_name="Customer",
        wait_until_answered=True,
    )
)

from livekit import api

req = api.RoomCompositeEgressRequest(
    room_name="support-1",
    layout="speaker",
    audio_only=True,
    file_outputs=[api.EncodedFileOutput(
        file_type=api.EncodedFileType.OGG,
        filepath="recordings/{room_name}-{time}.ogg",
        s3=api.S3Upload(bucket="calls", region="eu-central-1", access_key="<key>", secret="<secret>"),
    )],
)
info = await lk.egress.start_room_composite_egress(req)       # then lk.egress.stop_egress(api.StopEgressRequest(egress_id=info.egress_id))

# egress.yaml
api_key: APIexample
api_secret: <secret>
ws_url: ws://livekit:7880
redis:
  address: redis:6379
health_port: 8081
prometheus_port: 9091
s3: { access_key: <key>, secret: <secret>, region: eu-central-1, bucket: calls }   # or gcp: / azure:

docker run --rm -e EGRESS_CONFIG_FILE=/out/config.yaml -v ~/livekit-egress:/out livekit/egress

ing = await lk.ingress.create_ingress(
    api.CreateIngressRequest(
        input_type=api.IngressInput.RTMP_INPUT, name="stream",
        room_name="live-1", participant_identity="streamer",
    )
)
print(ing.url, ing.stream_key)

services:
  redis:
    image: redis:7-alpine
    restart: unless-stopped
    ports: ["127.0.0.1:6379:6379"]

  livekit:
    image: livekit/livekit-server:v1.13.10
    command: --config /etc/livekit.yaml
    restart: unless-stopped
    network_mode: host
    environment:
      LIVEKIT_KEYS: "${LIVEKIT_API_KEY}: ${LIVEKIT_API_SECRET}"
    volumes: ["./livekit.yaml:/etc/livekit.yaml:ro"]
    depends_on: [redis]

  agent:
    build: ./agent
    restart: unless-stopped
    stop_grace_period: 10m
    environment:
      LIVEKIT_URL: ws://host.docker.internal:7880
      LIVEKIT_API_KEY: ${LIVEKIT_API_KEY}
      LIVEKIT_API_SECRET: ${LIVEKIT_API_SECRET}
    extra_hosts: ["host.docker.internal:host-gateway"]

FROM python:3.13-slim
ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
RUN useradd --create-home agent
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt          # livekit-agents[openai,silero,turn-detector]~=1.8
COPY agent.py .
USER agent
RUN python -m livekit.agents download-files
EXPOSE 8081
CMD ["python", "agent.py", "start"]

helm repo add livekit https://helm.livekit.io
helm install livekit livekit/livekit-server --namespace livekit --values values.yaml

# values.yaml (based on server-sample.yaml in livekit/livekit-helm)
replicaCount: 2
terminationGracePeriodSeconds: 18000          # chart default: 5 h, so old pods drain
livekit:
  log_level: info
  rtc: { use_external_ip: true, port_range_start: 50000, port_range_end: 60000, tcp_port: 7881 }
  redis: { address: redis.livekit.svc:6379 }
  keys: { myapikey: "<secret from a Secret manager>" }
  turn: { enabled: true, domain: turn.example.com, tls_port: 5349, udp_port: 3478, secretName: turn-tls, serviceType: LoadBalancer }
autoscaling: { enabled: true, minReplicas: 2, maxReplicas: 5, targetCPUUtilizationPercentage: 60 }

# agent Deployment (excerpt)
spec:
  template:
    spec:
      terminationGracePeriodSeconds: 1800        # longer than your longest call
      containers:
        - name: agent
          image: registry.example.com/voice-agent:1.4.0
          ports: [{ containerPort: 8081 }]
          readinessProbe: { httpGet: { path: /, port: 8081 } }
          resources: { requests: { cpu: "4", memory: 8Gi }, limits: { memory: 8Gi } }
          envFrom: [{ secretRef: { name: livekit-credentials } }]

from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

from livekit.agents import JobContext
from livekit.agents.telemetry import set_tracer_provider


@server.rtc_session(agent_name="support", on_session_end=on_session_end)
async def entrypoint(ctx: JobContext) -> None:
    provider = TracerProvider()
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))        # endpoint from OTEL_EXPORTER_OTLP_ENDPOINT
    set_tracer_provider(provider, metadata={"session.room": ctx.room.name})    # attributes added to every span

    async def flush() -> None:
        provider.force_flush()                                                  # flush spans before the process exits

    ctx.add_shutdown_callback(flush)
    ...