This question originally came up in our Slack community and the thread has been consolidated here for long-term reference.
What are the best practices for passing around a LiveKitAPI object in a large Python application?
Would placing it in a container class with static accessors and mutex locks suffice? How can I ensure it gets properly closed when the process exits?
I’m troubleshooting some memory leaks and this is a potential suspect.
Both async with LiveKitAPI() as api (for one-off use) and caching it for later use are fine approaches.
In the job context, it’s defined as a cached property:
@functools.cached_property
def api(self) -> api.LiveKitAPI:
return api.LiveKitAPI(session=http_context.http_session())
If you use your own aiohttp.ClientSession instance as the session and close it properly, api.aclose() is not needed. The aclose() method just handles closing the aiohttp session object.
For agents, you can access the API from job_ctx.api:
if recorder_io:
if recorder_io.output_path:
sr.audio_recording_path = recorder_io.output_path
if recorder_io.recording_started_at:
sr.audio_recording_started_at = recorder_io.recording_started_at
sr.duration = sr.timestamp - sr.audio_recording_started_at
return sr
@functools.cached_property
def api(self) -> api.LiveKitAPI:
"""Returns an LiveKitAPI for making API calls to LiveKit.
Credentials are sourced from environment variables if not provided explicitly.
When starting via the worker, values passed in `WorkerOptions` are exported to
LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET so this API is always
usable inside job entrypoints.
"""
return api.LiveKitAPI(session=http_context.http_session())
@property