Job lifecycle

async def do_something(track: rtc.RemoteAudioTrack):
    audio_stream = rtc.AudioStream(track)
    async for event in audio_stream:
        # Do something here to process event.frame
        pass
    await audio_stream.aclose()

@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: JobContext):
    # an rtc.Room instance from the LiveKit Python SDK
    room = ctx.room

    # set up listeners on the room before connecting
    @room.on("track_subscribed")
    def on_track_subscribed(track: rtc.Track, *_):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            asyncio.create_task(do_something(track))

    # connect to room
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    # when connected, room.local_participant represents the agent
    await room.local_participant.send_text('hello world', topic='hello-world')

    # iterate through currently connected remote participants
    for rp in room.remote_participants.values():
        print(rp.identity)

async function doSomething(track: RemoteTrack) {
  for await (const frame of new AudioStream(track)) {
    // do something with the frame
  }
}

export default defineAgent({
  entry: async (ctx: JobContext) => {
    // an rtc.Room instance from the LiveKit Node.js SDK
    const room = ctx.room;

    // set up listeners on the room before connecting
    room.on(RoomEvent.TrackSubscribed, async (track: RemoteTrack) => {
      if (track.kind === TrackKind.KIND_AUDIO) {
        doSomething(track);
      }
    });

    await ctx.connect(undefined, AutoSubscribe.AUDIO_ONLY);

    // when connected, room.localParticipant represents the agent
    await room.localParticipant?.sendText('hello world', {
      topic: 'hello-world',
    });

    // iterate through currently connected remote participants
    for (const rp of ctx.room.remoteParticipants.values()) {
      console.log(rp.identity);
    }
  },
});

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    async def greet_participant(ctx: JobContext, p: rtc.RemoteParticipant):
        # Access participant identity, attributes, and metadata
        logger.info(f"participant joined: {p.identity}")

        # Filter out participants you don't need to handle
        if p.identity == "some-service-bot":
            return

        # Run participant-scoped work
        await ctx.room.local_participant.send_text(
            f"Hello, {p.identity}!", topic="greeting"
        )

    async def track_participant_audio(ctx: JobContext, p: rtc.RemoteParticipant):
        # Multiple entrypoints run concurrently for each participant
        logger.info(f"tracking audio for {p.identity}")
        await asyncio.sleep(60)

    # Register entrypoints before connecting
    ctx.add_participant_entrypoint(entrypoint_fnc=greet_participant)
    ctx.add_participant_entrypoint(entrypoint_fnc=track_participant_audio)

    await ctx.connect(auto_subscribe=AutoSubscribe.SUBSCRIBE_ALL)

export default defineAgent({
  entry: async (ctx: JobContext) => {
    const greetParticipant = async (ctx: JobContext, p: RemoteParticipant) => {
      // Access participant identity, attributes, and metadata
      console.log(`participant joined: ${p.identity}`);

      // Filter out participants you don't need to handle
      if (p.identity === 'some-service-bot') {
        return;
      }

      // Run participant-scoped work
      await ctx.room.localParticipant?.sendText(
        `Hello, ${p.identity}!`, { topic: 'greeting' }
      );
    };

    const trackParticipantAudio = async (ctx: JobContext, p: RemoteParticipant) => {
      // Multiple entrypoints run concurrently for each participant
      console.log(`tracking audio for ${p.identity}`);
      await new Promise((resolve) => setTimeout(resolve, 60_000));
    };

    // Register entrypoints before connecting
    ctx.addParticipantEntrypoint(greetParticipant);
    ctx.addParticipantEntrypoint(trackParticipantAudio);

    await ctx.connect(undefined, AutoSubscribe.SUBSCRIBE_ALL);
  },
});

import json

@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: JobContext):
    metadata = json.loads(ctx.job.metadata)
    user_id = metadata["user_id"]
    user_name = metadata["user_name"]
    user_phone = metadata["user_phone"]
    # ...

export default defineAgent({
  entry: async (ctx: JobContext) => {
    const metadata = JSON.parse(ctx.job.metadata);
    const userId = metadata.user_id;
    const userName = metadata.user_name;
    const userPhone = metadata.user_phone;
    // ...
  },
});

@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: JobContext):
  # connect to the room
  await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

  # wait for the first participant to arrive
  participant = await ctx.wait_for_participant()

  # customize behavior based on the participant
  print(f"connected to room {ctx.room.name} with participant {participant.identity}")

  # inspect the current value of the attribute
  language = participant.attributes.get("user.language")

  # listen to when the attribute is changed
  @ctx.room.on("participant_attributes_changed")
  def on_participant_attributes_changed(changed_attrs: dict[str, str], p: rtc.Participant):
      if p == participant:
        language = p.attributes.get("user.language")
        print(f"participant {p.identity} changed language to {language}")

export default defineAgent({
  entry: async (ctx: JobContext) => {
    // connect to the room
    await ctx.connect(undefined, AutoSubscribe.AUDIO_ONLY);

    // wait for the first participant to arrive
    const participant = await ctx.waitForParticipant();

    // customize behavior based on the participant
    console.log(`connected to room ${ctx.room.name} with participant ${participant.identity}`);

    // inspect the current value of the attribute
    let language = participant.attributes['user.language'];

    // listen to when the attribute is changed
    ctx.room.on(
      'participantAttributesChanged',
      (changedAttrs: Record<string, string>, p: Participant) => {
        if (p === participant) {
          language = p.attributes['user.language'];
          console.log(`participant ${p.identity} changed language to ${language}`);
        }
      },
    );
  },
});

# Graceful shutdown with draining
session.shutdown(drain=True)

# Or immediate close
await session.aclose()

export default defineAgent({
  entry: async (ctx: JobContext) => {
    // do some work...

    // Graceful shutdown with draining
    ctx.shutdown(drain=true);

    // Or immediate close
    await ctx.aclose();
  },
});

from livekit import api

async def entrypoint(ctx: JobContext):
    # do some work
    ...

    api_client = api.LiveKitAPI(
        os.environ["LIVEKIT_URL"],
        os.environ["LIVEKIT_API_KEY"],
        os.environ["LIVEKIT_API_SECRET"],
    )
    await api_client.room.delete_room(api.DeleteRoomRequest(
        room=ctx.job.room.name,
    ))

export default defineAgent({
  entry: async (ctx: JobContext) => {
    // do some work...

    const roomServiceClient = new RoomServiceClient(
      process.env.LIVEKIT_URL,
      process.env.LIVEKIT_API_KEY,
      process.env.LIVEKIT_API_SECRET,
    );
    await roomServiceClient.deleteRoom(ctx.job.room.name);
  },
});

async def entrypoint(ctx: JobContext):
    async def my_shutdown_hook():
        # save user state
        ...
    ctx.add_shutdown_callback(my_shutdown_hook)

export default defineAgent({
  entry: async (ctx: JobContext) => {
    ctx.addShutdownCallback(() => {
      // save user state...
    });
  },
});