How to get the exact count of active jobs for the worker at that moment in request_fnc?

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

current_job_count = len(req.job.agent._agent_server.active_jobs) Gives attribute error len(req.job.agent._agent_server.active_jobs)\n ^^^^^^^^^^^^^\nAttributeError: agent"

Thanks — I’m responding to the AttributeError when you try

len(req.job.agent._agent_server.active_jobs)

. That AttributeError happens because

req.job

does not have an

agent

attribute. You should read the AgentServer directly (or use a Server reference in scope) to get the active jobs count. If your AgentServer instance is named

server

, do this inside `request_fnc`:

async def request_fnc(req: JobRequest):
    current_job_count = len(server.active_jobs)

    if current_job_count >= 10:
        await req.reject()
    else:
        await req.accept()

If

server

is not accessible in that scope, pass a reference into the request function (closure or attribute) or store it where

request_fnc

can see it. If you’d like, share a minimal snippet showing how you create/register your AgentServer and

request_fnc

and I’ll point out the exact place to reference

server

. Sources: Agent server options | LiveKit docs | Job lifecycle | LiveKit docs

Agent is run like this cli.run_app( WorkerOptions( api_key=settings.livekit_api_key, api_secret=settings.livekit_api_secret, ws_url=settings.livekit_url, entrypoint_fnc=entrypoint, agent_name=settings.agent_name, request_fnc=request_fnc, drain_timeout=drain_timeout, shutdown_process_timeout=shutdown_process_timeout, load_fnc=calc_and_record_load, load_threshold=settings.worker_load_threshold or 0.7, ) ) async def request_fnc(req: JobRequest):

Thanks — you're running the agent with WorkerOptions and

request_fnc

as a standalone function. The AttributeError happens because

req.job

does not expose an `agent`/`_agent_server` reference. You need to read

active_jobs

from the AgentServer instance that actually tracks them. Two simple fixes — pick one that fits your structure: Option 1 — capture the server in a closure

server = AgentServer(...)

async def request_fnc(req: JobRequest):
    current_job_count = len(server.active_jobs)
    if current_job_count >= 10:
        await req.reject()
    else:
        await req.accept()

Then pass that

request_fnc

into WorkerOptions. Option 2 — create a factory that binds the server

def make_request_fnc(server):
    async def request_fnc(req: JobRequest):
        current_job_count = len(server.active_jobs)
        if current_job_count >= 10:
            await req.reject()
        else:
            await req.accept()
    return request_fnc

server = AgentServer(...)
request_fnc = make_request_fnc(server)

Key points: • Use

len(server.active_jobs)

—

req.job.agent._agent_server

is not available in

request_fnc

. • Ensure the

server

variable is the same AgentServer instance used to run the worker (so its

active_jobs

reflects real state). If you want, paste the minimal snippet where you create

server

and call

cli.run_app(...)

and I’ll show exactly where to bind

request_fnc

. Sources: Agent server options | LiveKit docs | Server lifecycle | LiveKit docs | Custom agent deployments | LiveKit docs

How to change from worker options to server give my current code.

cli.run_app(
        WorkerOptions(
            api_key=settings.livekit_api_key,
            api_secret=settings.livekit_api_secret,
            ws_url=settings.livekit_url,
            entrypoint_fnc=entrypoint,
            agent_name=settings.agent_name,
            request_fnc=request_fnc,
            drain_timeout=drain_timeout,
            shutdown_process_timeout=shutdown_process_timeout,
            load_fnc=calc_and_record_load,
            load_threshold=settings.worker_load_threshold or 0.7,
        )
    )

You're switching from the WorkerOptions helper to creating and running an AgentServer so

request_fnc

can access

server.active_jobs

. Below is a minimal refactor that keeps your same settings and hooks but exposes the server instance to

request_fnc

. 1) Create the server, bind your handlers, then run it.

from livekit.agents import AgentServer, JobRequest
# import other needed names (cli, settings, etc.)

# Create server instance with the same options you were passing
server = AgentServer(
    api_key=settings.livekit_api_key,
    api_secret=settings.livekit_api_secret,
    ws_url=settings.livekit_url,
    agent_name=settings.agent_name,
    drain_timeout=drain_timeout,
    shutdown_process_timeout=shutdown_process_timeout,
    load_fnc=calc_and_record_load,
    load_threshold=settings.worker_load_threshold or 0.7,
)

# request_fnc can now reference `server` (closure)
async def request_fnc(req: JobRequest):
    current_job_count = len(server.active_jobs)
    if current_job_count >= 10:
        await req.reject()
    else:
        await req.accept()

# Attach entrypoint and request handler (use decorator or explicit registration)
@server.rtc_session(on_request=request_fnc)
async def entrypoint(ctx):
    # your existing agent logic
    pass

# Start the server with cli.run_app
if __name__ == "__main__":
    cli.run_app(server)

Notes: •

len(server.active_jobs)

gives the exact current active job count. • Ensure you import the same symbols

cli

expects and use the same

entrypoint

logic. • If you prefer not to use a closure, use a factory to bind the server:

def make_request_fnc(server):
    async def request_fnc(req: JobRequest):
        current_job_count = len(server.active_jobs)
        ...
    return request_fnc

request_fnc = make_request_fnc(server)

If you share your current

entrypoint

and where

cli

is imported from, I can give a tailored snippet with exact imports and any missing pieces.

Is len(server.active_jobs) reliable given the async behavior of the running agent?

You're asking whether

len(server.active_jobs)

is reliable given the agent's asynchronous behavior — good question. Yes —

server.active_jobs

is the AgentServer's runtime list of currently executing jobs and

len(server.active_jobs)

reflects the current count at the moment you read it. The framework updates that list as jobs start and finish, so it's the correct way to make load/accept decisions in

request_fnc

. Caveats and best practices: • It's a snapshot: because your code and the server run concurrently, the count can change immediately after you read it (race condition). Treat the value as the current state, not a locked guarantee. • Keep decision logic short and immediate: perform accept/reject quickly after reading the count to minimize races. • Use conservative thresholds: if exact strict limits matter, subtract a safety margin (e.g., allow up to N-1) or reserve capacity for in-flight accepts. • If you need strong atomic guarantees (no other accept can occur between check and accept), implement an explicit capacity reservation or centralized coordination rather than relying solely on the snapshot. If you want, tell me your desired strictness (soft limit vs absolute max) and I’ll suggest a concrete pattern (safety margin, reservation token, or queueing) you can implement.

Is the reject method guaranteed to work?

You're asking whether

req.reject()

is guaranteed to prevent a job from being accepted/processed. Yes — calling

await req.reject()

in your

request_fnc

reliably rejects that job request so the AgentServer will not accept or run it. The framework treats a rejection as the agent declining the job and will route the job to other agents (or leave it unassigned per system logic). Caveats to keep in mind: •

reject()

is final for that agent: once called, that agent will not process the job. • Timing/races:

reject()

prevents this agent from accepting the job at the moment you call it. Ensure you call it quickly after your load check — the system relies on your

on_request

decision being made before any accept occurs. • Default behavior: if you don’t implement

request_fnc

(or don’t call

reject

), the default is to accept requests, so explicit rejects are necessary when you want to decline. If you need stronger guarantees against race conditions (e.g., multiple agents checking capacity simultaneously), I can suggest patterns such as a small reservation token, a conservative threshold, or a centralized capacity check — tell me which fit your setup and I’ll provide code. Sources: Agent server options | LiveKit docs

server = AgentServer( api_key=settings.livekit_api_key, api_secret=settings.livekit_api_secret, ws_url=settings.livekit_url, drain_timeout=WORKER_DRAIN_TIMEOUT, shutdown_process_timeout=WORKER_PROCESS_SHUTDOWN_TIMEOUT, load_fnc=calc_and_record_load, load_threshold=settings.worker_load_threshold or 0.7, ) server.rtc_session( on_request=make_request_fnc(server), agent_name=settings.agent_name, ) async def entrypoint(ctx: JobContext): logger.info("Total jobs are", count=len(server.active_jobs)) nAttributeError: 'AgentServer' object has no attribute '_proc_pool'"

Thanks — I’ll respond to the AttributeError and show the minimal correct pattern to run an AgentServer so

server.active_jobs

is usable at runtime. Problem • The AttributeError

'AgentServer' object has no attribute '_proc_pool'

happens because internal server state (like

_proc_pool

and

active_jobs

) is only initialized when the server is started. Accessing

server.active_jobs

at import/definition time (before

cli.run_app(server)

runs) triggers this error. Fix — minimal refactor 1. Create the AgentServer instance. 2. Bind handlers (use a closure or factory so

request_fnc

can see

server

). 3. Register the entrypoint with

server.rtc_session(...)(entrypoint)

. 4. Start with

cli.run_app(server)

so internal state is initialized. Example:

from livekit.agents import AgentServer, JobContext
# build server (do not pass api_key/api_secret here if your environment provides them)
server = AgentServer(
    drain_timeout=WORKER_DRAIN_TIMEOUT,
    shutdown_process_timeout=WORKER_PROCESS_SHUTDOWN_TIMEOUT,
    load_fnc=calc_and_record_load,
    load_threshold=settings.worker_load_threshold or 0.7,
)

def make_request_fnc(server):
    async def request_fnc(req):
        # runtime access to active_jobs is safe
        current_job_count = len(server.active_jobs)
        if current_job_count >= 10:
            await req.reject()
        else:
            await req.accept()
    return request_fnc

# Bind request handler and entrypoint (decorator form)
server.rtc_session(
    on_request=make_request_fnc(server),
    agent_name=settings.agent_name,
)(entrypoint)

if __name__ == "__main__":
    cli.run_app(server)

Key notes • Only access

server.active_jobs

at runtime (inside

request_fnc

,

entrypoint

, or after

cli.run_app(server)

has started). Do not read it at module/import time. •

await req.reject()

reliably prevents this agent from accepting the job. It’s final for this agent. •

len(server.active_jobs)

is a snapshot; it can change immediately after you read it. Use conservative thresholds or a reservation pattern if you need stricter guarantees. If you want, paste your current

entrypoint

and where you log

len(server.active_jobs)

and I’ll show the exact lines to move so the AttributeError disappears. Sources: Agent server options | LiveKit docs

can livekit agent server read log level directly from env or specified in the code?

Short answer: LiveKit AgentServer does not automatically read a log level from the environment — you must set logging yourself (or pass a CLI flag when starting the app). You can do that either in code (recommended) or via the CLI. Two simple options: • Set log level from env in code

import os, logging
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=log_level)

Place this before creating the AgentServer so all logs use that level. • Pass a CLI flag when starting Run your app with the

--log-level

flag:

python agent.py dev --log-level=DEBUG

or your equivalent

cli.run_app

invocation that accepts

--log-level

. If you tell me how you start the process (direct Python, systemd, container), I can suggest the exact place to add the env var or the recommended logging setup. Sources: Agent server options | LiveKit docs | Simple Call Answering Agent | LiveKit docs