livekit-server-sdk

name: livekit-server-sdk description: Use LiveKit Server SDKs for Python, Node.js, and Go to manage rooms, participants, tokens, and backend operations argument-hint: "" allowed-tools: Read, Write, Bash(npm install, pip install, go get), Glob, Grep

LiveKit Server SDKs

Backend operations with LiveKit: $ARGUMENTS

Expert Knowledge

You are a LiveKit server SDK specialist with expertise in:

• Token generation
• Room management
• Participant control
• Track management
• Webhook handling
• Multi-language SDK usage

SDK Overview

Python SDK

Installation

pip install livekit-api

Token Generation

from livekit.api import AccessToken, VideoGrant
from datetime import timedelta
import os

def generate_token(room_name: str, identity: str) -> str:
    token = AccessToken(
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )
    token.identity = identity
    token.ttl = timedelta(hours=2)

    grant = VideoGrant(
        room_join=True,
        room=room_name,
        can_publish=True,
        can_subscribe=True,
    )
    token.add_grant(grant)

    return token.to_jwt()

Room Management

from livekit.api import LiveKitAPI
from livekit.protocol import room as room_proto

api = LiveKitAPI()

# Create room
room = await api.room.create_room(
    room_proto.CreateRoomRequest(
        name="my-room",
        empty_timeout=600,
        max_participants=50,
        metadata='{"type": "meeting"}',
    )
)

# List rooms
rooms = await api.room.list_rooms(
    room_proto.ListRoomsRequest()
)

# Delete room
await api.room.delete_room(
    room_proto.DeleteRoomRequest(room="my-room")
)

# Update room metadata
await api.room.update_room_metadata(
    room_proto.UpdateRoomMetadataRequest(
        room="my-room",
        metadata='{"status": "active"}',
    )
)

Participant Management

# List participants
participants = await api.room.list_participants(
    room_proto.ListParticipantsRequest(room="my-room")
)

# Get participant
participant = await api.room.get_participant(
    room_proto.RoomParticipantIdentity(
        room="my-room",
        identity="user-123",
    )
)

# Update participant
await api.room.update_participant(
    room_proto.UpdateParticipantRequest(
        room="my-room",
        identity="user-123",
        metadata='{"role": "moderator"}',
        permission=room_proto.ParticipantPermission(
            can_publish=True,
            can_subscribe=True,
        ),
    )
)

# Remove participant
await api.room.remove_participant(
    room_proto.RoomParticipantIdentity(
        room="my-room",
        identity="user-123",
    )
)

# Mute track
await api.room.mute_published_track(
    room_proto.MuteRoomTrackRequest(
        room="my-room",
        identity="user-123",
        track_sid="TR_xxx",
        muted=True,
    )
)

Send Data

await api.room.send_data(
    room_proto.SendDataRequest(
        room="my-room",
        data=b'{"type": "notification", "message": "Hello!"}',
        kind=room_proto.DataPacket.RELIABLE,
        destination_identities=["user-1", "user-2"],  # Optional
    )
)

Node.js SDK

Installation

npm install livekit-server-sdk

Token Generation

import { AccessToken } from 'livekit-server-sdk';

function generateToken(roomName: string, identity: string): string {
  const token = new AccessToken(
    process.env.LIVEKIT_API_KEY!,
    process.env.LIVEKIT_API_SECRET!,
    {
      identity,
      ttl: '2h',
    }
  );

  token.addGrant({
    roomJoin: true,
    room: roomName,
    canPublish: true,
    canSubscribe: true,
  });

  return token.toJwt();
}

Room Service

import { RoomServiceClient } from 'livekit-server-sdk';

const roomService = new RoomServiceClient(
  process.env.LIVEKIT_URL!,
  process.env.LIVEKIT_API_KEY!,
  process.env.LIVEKIT_API_SECRET!
);

// Create room
await roomService.createRoom({
  name: 'my-room',
  emptyTimeout: 600,
  maxParticipants: 50,
  metadata: JSON.stringify({ type: 'meeting' }),
});

// List rooms
const rooms = await roomService.listRooms();

// Delete room
await roomService.deleteRoom('my-room');

// List participants
const participants = await roomService.listParticipants('my-room');

// Update participant
await roomService.updateParticipant('my-room', 'user-123', {
  metadata: JSON.stringify({ role: 'moderator' }),
  permission: {
    canPublish: true,
    canSubscribe: true,
  },
});

// Remove participant
await roomService.removeParticipant('my-room', 'user-123');

// Mute track
await roomService.mutePublishedTrack('my-room', 'user-123', 'TR_xxx', true);

// Send data
await roomService.sendData(
  'my-room',
  Buffer.from(JSON.stringify({ type: 'notification' })),
  { kind: 'RELIABLE', destinationIdentities: ['user-1'] }
);

Egress Service

import { EgressClient } from 'livekit-server-sdk';

const egress = new EgressClient(
  process.env.LIVEKIT_URL!,
  process.env.LIVEKIT_API_KEY!,
  process.env.LIVEKIT_API_SECRET!
);

// Start recording
const recording = await egress.startRoomCompositeEgress('my-room', {
  file: {
    filepath: 'recordings/{room_name}/{time}.mp4',
    output: {
      s3: {
        bucket: 'my-bucket',
        region: 'us-east-1',
        accessKey: process.env.AWS_ACCESS_KEY!,
        secret: process.env.AWS_SECRET_KEY!,
      },
    },
  },
});

// Stop recording
await egress.stopEgress(recording.egressId);

// List egresses
const egresses = await egress.listEgress({ roomName: 'my-room' });

Webhook Receiver

import { WebhookReceiver } from 'livekit-server-sdk';
import express from 'express';

const app = express();
app.use('/webhook', express.raw({ type: 'application/webhook+json' }));

const receiver = new WebhookReceiver(
  process.env.LIVEKIT_API_KEY!,
  process.env.LIVEKIT_API_SECRET!
);

app.post('/webhook', async (req, res) => {
  try {
    const event = await receiver.receive(req.body, req.get('Authorization'));
    console.log('Event:', event.event);

    // Handle event...

    res.sendStatus(200);
  } catch (error) {
    res.sendStatus(400);
  }
});

Go SDK

Installation

go get github.com/livekit/server-sdk-go/v2

Token Generation

package main

import (
    "os"
    "time"

    "github.com/livekit/protocol/auth"
)

func generateToken(roomName, identity string) (string, error) {
    apiKey := os.Getenv("LIVEKIT_API_KEY")
    apiSecret := os.Getenv("LIVEKIT_API_SECRET")

    at := auth.NewAccessToken(apiKey, apiSecret)
    grant := &auth.VideoGrant{
        RoomJoin: true,
        Room:     roomName,
    }
    at.AddGrant(grant).
        SetIdentity(identity).
        SetValidFor(2 * time.Hour)

    return at.ToJWT()
}

Room Service

package main

import (
    "context"
    "os"

    lksdk "github.com/livekit/server-sdk-go/v2"
    "github.com/livekit/protocol/livekit"
)

func main() {
    host := os.Getenv("LIVEKIT_URL")
    apiKey := os.Getenv("LIVEKIT_API_KEY")
    apiSecret := os.Getenv("LIVEKIT_API_SECRET")

    roomClient := lksdk.NewRoomServiceClient(host, apiKey, apiSecret)
    ctx := context.Background()

    // Create room
    room, err := roomClient.CreateRoom(ctx, &livekit.CreateRoomRequest{
        Name:            "my-room",
        EmptyTimeout:    600,
        MaxParticipants: 50,
    })

    // List rooms
    rooms, err := roomClient.ListRooms(ctx, &livekit.ListRoomsRequest{})

    // Delete room
    _, err = roomClient.DeleteRoom(ctx, &livekit.DeleteRoomRequest{
        Room: "my-room",
    })

    // List participants
    participants, err := roomClient.ListParticipants(ctx, &livekit.ListParticipantsRequest{
        Room: "my-room",
    })

    // Remove participant
    _, err = roomClient.RemoveParticipant(ctx, &livekit.RoomParticipantIdentity{
        Room:     "my-room",
        Identity: "user-123",
    })
}

Express API Example

import express from 'express';
import {
  AccessToken,
  RoomServiceClient,
  EgressClient,
  WebhookReceiver,
} from 'livekit-server-sdk';

const app = express();
app.use(express.json());

const apiKey = process.env.LIVEKIT_API_KEY!;
const apiSecret = process.env.LIVEKIT_API_SECRET!;
const livekitUrl = process.env.LIVEKIT_URL!;

const roomService = new RoomServiceClient(livekitUrl, apiKey, apiSecret);
const egress = new EgressClient(livekitUrl, apiKey, apiSecret);
const webhookReceiver = new WebhookReceiver(apiKey, apiSecret);

// Generate token
app.post('/api/token', async (req, res) => {
  const { roomName, identity, permissions } = req.body;

  const token = new AccessToken(apiKey, apiSecret, {
    identity,
    ttl: '2h',
  });

  token.addGrant({
    roomJoin: true,
    room: roomName,
    canPublish: permissions?.publish ?? true,
    canSubscribe: permissions?.subscribe ?? true,
  });

  res.json({ token: token.toJwt() });
});

// Create room
app.post('/api/rooms', async (req, res) => {
  const { name, maxParticipants } = req.body;

  const room = await roomService.createRoom({
    name,
    maxParticipants,
    emptyTimeout: 600,
  });

  res.json(room);
});

// List rooms
app.get('/api/rooms', async (req, res) => {
  const rooms = await roomService.listRooms();
  res.json(rooms);
});

// Delete room
app.delete('/api/rooms/:name', async (req, res) => {
  await roomService.deleteRoom(req.params.name);
  res.sendStatus(204);
});

// List participants
app.get('/api/rooms/:name/participants', async (req, res) => {
  const participants = await roomService.listParticipants(req.params.name);
  res.json(participants);
});

// Kick participant
app.delete('/api/rooms/:name/participants/:identity', async (req, res) => {
  await roomService.removeParticipant(req.params.name, req.params.identity);
  res.sendStatus(204);
});

// Start recording
app.post('/api/rooms/:name/recording', async (req, res) => {
  const recording = await egress.startRoomCompositeEgress(req.params.name, {
    file: {
      filepath: `recordings/${req.params.name}/{time}.mp4`,
    },
  });

  res.json({ egressId: recording.egressId });
});

// Stop recording
app.delete('/api/recording/:egressId', async (req, res) => {
  await egress.stopEgress(req.params.egressId);
  res.sendStatus(204);
});

// Webhook handler
app.post('/webhook', express.raw({ type: 'application/webhook+json' }), async (req, res) => {
  try {
    const event = await webhookReceiver.receive(req.body, req.get('Authorization'));

    switch (event.event) {
      case 'room_started':
        console.log('Room started:', event.room?.name);
        break;
      case 'participant_joined':
        console.log('Participant joined:', event.participant?.identity);
        break;
      // Handle other events...
    }

    res.sendStatus(200);
  } catch (error) {
    res.sendStatus(400);
  }
});

app.listen(3000);

FastAPI Example

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from livekit.api import LiveKitAPI, AccessToken, VideoGrant, WebhookReceiver
from livekit.protocol import room as room_proto
from datetime import timedelta
import os

app = FastAPI()
api = LiveKitAPI()
webhook_receiver = WebhookReceiver(
    os.environ["LIVEKIT_API_KEY"],
    os.environ["LIVEKIT_API_SECRET"],
)

class TokenRequest(BaseModel):
    room_name: str
    identity: str
    can_publish: bool = True
    can_subscribe: bool = True

class RoomRequest(BaseModel):
    name: str
    max_participants: int = 50

@app.post("/api/token")
async def generate_token(request: TokenRequest):
    token = AccessToken(
        os.environ["LIVEKIT_API_KEY"],
        os.environ["LIVEKIT_API_SECRET"],
    )
    token.identity = request.identity
    token.ttl = timedelta(hours=2)
    token.add_grant(VideoGrant(
        room_join=True,
        room=request.room_name,
        can_publish=request.can_publish,
        can_subscribe=request.can_subscribe,
    ))

    return {"token": token.to_jwt()}

@app.post("/api/rooms")
async def create_room(request: RoomRequest):
    room = await api.room.create_room(
        room_proto.CreateRoomRequest(
            name=request.name,
            max_participants=request.max_participants,
            empty_timeout=600,
        )
    )
    return {"room": room}

@app.get("/api/rooms")
async def list_rooms():
    rooms = await api.room.list_rooms(room_proto.ListRoomsRequest())
    return {"rooms": rooms}

@app.delete("/api/rooms/{name}")
async def delete_room(name: str):
    await api.room.delete_room(room_proto.DeleteRoomRequest(room=name))
    return {"status": "deleted"}

@app.post("/webhook")
async def handle_webhook(request: Request):
    body = await request.body()
    auth = request.headers.get("Authorization")

    try:
        event = webhook_receiver.receive(body, auth)
        print(f"Event: {event.event}")
        return {"status": "ok"}
    except Exception as e:
        raise HTTPException(400, str(e))

Best Practices

1. Secure API secrets: Never expose in client code
2. Validate inputs: Check room/identity before operations
3. Handle errors: Wrap API calls in try/catch
4. Use webhooks: React to events instead of polling
5. Set timeouts: Prevent hanging connections

Deliverables

For: $ARGUMENTS

Provide:

1. SDK installation
2. Token generation endpoint
3. Room management API
4. Participant control
5. Webhook handling
6. Error handling