async def stop_room_recording(self, egress_id: str) -> bool:
    """Stop a room composite egress recording.

    Args:
        egress_id: The egress ID to stop

    Returns:
        True if successful, False otherwise
    """
    async with api.LiveKitAPI(
        self.livekit_url,
        self.livekit_api_key,
        self.livekit_api_secret,
    ) as lkapi:
        stop_req: StopEgressRequest = api.StopEgressRequest(egress_id=egress_id)
        await lkapi.egress.stop_egress(stop_req)
        <http://logger.info|logger.info>(f"Stopped room composite egress: egress_id={egress_id}")
    return True

here i get 2025-12-15 125805,821 INFO AtlasBackend: Successfully uploaded transcript for session aab8dc23-8520-4b95-9138-327ae5d0cfb9 {"level": "INFO", "name": "AtlasBackend", "message": "Successfully uploaded transcript for session aab8dc23-8520-4b95-9138-327ae5d0cfb9", "pid": 33, "job_id": "AJ_6iWpmYyF9MhV", "room_id": "RM_mBzMp6niAnco", "timestamp": "2025-12-15T125805.821123+00:00"} {"level": "ERROR", "name": "livekit.agents", "message": "error while shutting down the job\nTraceback (most recent call last):\n File \"/app/.venv/lib/python3.12/site-packages/livekit/agents/ipc/job_proc_lazy_main.py\", line 351, in _run_job_task\n await asyncio.gather(*shutdown_tasks)\n File \"/app/.venv/lib/python3.12/site-packages/livekit/agents/job.py\", line 361, in wrapper\n await callback() # type: ignore\n ^^^^^^^^^^^^^^^^\n File \"/app/core/session/event_handlers.py\", line 167, in stop_egress\n success = await egress_service.stop_room_recording(session_state.egress_id)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/app/services/recording/egress_service.py\", line 82, in stop_room_recording\n await lkapi.egress.stop_egress(stop_req)\n File \"/app/.venv/lib/python3.12/site-packages/livekit/api/egress_service.py\", line 128, in stop_egress\n return await self._client.request(\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File \"/app/.venv/lib/python3.12/site-packages/livekit/api/twirp_client.py\", line 129, in request\n raise TwirpError(\nlivekit.api.twirp_client.TwirpError: TwirpError(code=failed_precondition, message=egress with status EGRESS_COMPLETE cannot be stopped, status=412)", "pid": 33, "job_id": "AJ_6iWpmYyF9MhV", "room_id": "RM_mBzMp6niAnco", "timestamp": "2025-12-15T125808.837374+00:00"}

Thanks for reaching out, Aleh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

is there some getegress by id method?

Short answer: No — there’s no dedicated

GetEgressById

method. You can get a specific egress’ info by calling

list_egress()

and filtering the returned

egress

list for your

egress_id

. That lets you inspect its

status

and other fields before calling

stop_egress

. If you want, I can: • show a concise code snippet that wraps

list_egress()

into a

get_egress_by_id(egress_id)

helper, or • review a

list_egress()

response you have and point out the fields to check. Which would you prefer?

async def stop_room_recording(self, egress_id: str) -> bool:
    """Stop a room composite egress recording.

    Args:
        egress_id: The egress ID to stop

    Returns:
        True if successful, False otherwise
    """
    async with api.LiveKitAPI(
        self.livekit_url,
        self.livekit_api_key,
        self.livekit_api_secret,
    ) as lkapi:
        list_res = await lkapi.egress.list_egress()
        eg = next((e for e in list_res.egress if e.egress_id == egress_id), None)
        if not eg.status == api.EgressStatus.EGRESS_COMPLETE:
            stop_req: StopEgressRequest = api.StopEgressRequest(egress_id=egress_id)
            await lkapi.egress.stop_egress(stop_req)
            <http://logger.info|logger.info>(f"Stopped room composite egress: egress_id={egress_id}")
        return True

TypeError: EgressService.list_egress() missing 1 required positional argument: 'list'

You're seeing that TypeError because

list_egress()

expects a request object argument, not an empty call. Pass a

ListEgressRequest

(even an empty one) and check for a missing

eg

before accessing

.status

. Also fix the status check logic —

if not eg.status == ...

is confusing; use explicit guards. Concise corrected example:

from livekit import api

async def stop_room_recording(self, egress_id: str) -> bool:
    async with api.LiveKitAPI(self.livekit_url, self.livekit_api_key, self.livekit_api_secret) as lkapi:
        # pass a request object (can be empty)
        list_req = api.ListEgressRequest()
        list_res = await lkapi.egress.list_egress(list_req)

        eg = next((e for e in list_res.egress if e.egress_id == egress_id), None)
        if eg is None:
            logger.warning(f"Egress {egress_id} not found")
            return False

        if eg.status == api.EgressStatus.EGRESS_COMPLETE:
            <http://logger.info|logger.info>(f"Egress {egress_id} already complete")
            return True

        # stop only when not in a terminal state
        stop_req = api.StopEgressRequest(egress_id=egress_id)
        await lkapi.egress.stop_egress(stop_req)
        <http://logger.info|logger.info>(f"Stopped room composite egress: egress_id={egress_id}")
        return True

If you want, paste a sample

list_egress()

response and I’ll point out which fields you should inspect for additional safety (e.g.,

status

,

output

or timestamps).

req = api.RoomCompositeEgressRequest(
    room_name=room_name,
    audio_only=True,
    file_outputs=[
        api.EncodedFileOutput(
            filepath=filepath,
            file_type=file_type,
            azure=api.AzureBlobUpload(
                account_name=storage_settings.account_name,
                account_key=storage_settings.account_key,
                container_name=storage_settings.container_name,
            ),
        )
    ],
)

async with api.LiveKitAPI(
    self.livekit_url,
    self.livekit_api_key,
    self.livekit_api_secret,
) as lkapi:
    info: EgressInfo = await lkapi.egress.start_room_composite_egress(req)
    <http://logger.info|logger.info>(
        f"Started room composite egress: egress_id={info.egress_id}, room={room_name}"
    )
    return info.egress_id

this will auto complete on session close or not?

Short answer: No — RoomCompositeEgress does not automatically stop when your agent session closes. It only stops automatically when the entire room becomes empty (all participants have left). What you can do (pick one): • Stop by ID: call

stop_egress()

with the

egress_id

you got from

start_room_composite_egress

. • Delete the room: call

delete_room(room=room_name)

to disconnect everyone and terminate the egress. • Use webhooks: listen for

egress_ended

to update state and avoid duplicate stop calls. Recommended pattern (concise): 1. When you start egress, persist

egress_id

. 2. On your session shutdown, either call

stop_egress(egress_id)

or

delete_room(room_name)

. 3. Also subscribe to

egress_ended

so you won’t try to stop an already-completed egress. If you’d like, I can show a short code example for stopping the egress by ID or deleting the room. Which do you prefer? Sources: Composite and web recordings | LiveKit docs

hey @better-sandwich-65834 did u ever figure out the issue?

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!