load_fnc, what is the use of this def and how can ...
# ask-aib
boundless-author-7811
02/18/2026, 11:37 AMload_fnc, what is the use of this def and how can it be customized
t
tall-airline-16756
02/18/2026, 11:37 AMThanks for reaching out, Utkarsh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
02/18/2026, 11:38 AMThanks — you're asking what
load_fncload_fncload_fncload_fncload_thresholdload_fnc Copy code
def compute_load(agent_server: AgentServer) -> float:
# assume max desired concurrent jobs is 10
return min(len(agent_server.active_jobs) / 10, 1.0)
server = AgentServer(load_threshold=0.9)
server.load_fnc = compute_load Copy code
const computeLoad = async (agentServer: AgentServer): Promise<number> => {
// assume max desired concurrent jobs is 10
return Math.min(agentServer.activeJobs.length / 10, 1.0);
};
const server = new AgentServer({
loadFunc: computeLoad,
loadThreshold: 0.9,
});load_fncload_thresholdb
boundless-author-7811
02/18/2026, 12:19 PMi am not using AgentServer, i think this was a recent upgrade, and i am running the livekit agent in kubernetes and there are 2 pods running so how can i use the load_fnc with multiple pods running?
t
tall-airline-16756
02/18/2026, 12:19 PMThanks — that helps. You said you’re running the LiveKit agent in Kubernetes with two pods and you’re not using
AgentServerload_fncload_fncAgentServerAgentServerAgentServerload_fncload_fncAgentServerAgentServerload_fncload_thresholdload_fncload_fncload_fncAgentServerload_fncAgentServerb
boundless-author-7811
02/18/2026, 12:20 PMbelow is my entry point:
Copy code
import os
from livekit.agents import cli, WorkerOptions
from lifecycle import entrypoint, prewarm
from shared.utils.logger import get_logger
logger = get_logger("orchestrator.main")
# Prometheus metrics port (default: 9090, configurable via env)
PROMETHEUS_PORT = int(os.getenv("PROMETHEUS_PORT", "9090"))
if __name__ == "__main__":
<http://logger.info|logger.info>("🚀 Starting Orchestrator...")
<http://logger.info|logger.info>("📊 Prometheus metrics will be exposed on port %s", PROMETHEUS_PORT)
# Number of idle worker processes to keep warm
# CRITICAL: Must be 1 to prevent duplicate job dispatch to same room
# LiveKit dispatches room jobs to multiple workers if num_idle_processes > 1
# This causes multiple agents to join the same call
# NUM_IDLE_PROCESSES = int(os.getenv("NUM_IDLE_PROCESSES", "1"))
cli.run_app(
WorkerOptions(
entrypoint_fnc=entrypoint,
prewarm_fnc=prewarm,
agent_name="outbound-caller",
port=8001,
prometheus_port=PROMETHEUS_PORT
)
)t
tall-airline-16756
02/18/2026, 12:21 PMThanks — this is very helpful. You’re using the
WorkerOptionscli.run_appload_fnccli.run_app(WorkerOptions(...))load_fncload_fncAgentServerWorkerOptionscli.run_appAgentServerload_fncAgentServerload_fncload_thresholdAgentServerload_fncentrypointload_fncAgentServerload_fncWorkerOptionsload_fncentrypointb
boundless-author-7811
02/18/2026, 12:23 PMis the AgentServer a new implementation, when i started building the examples did not have AgentServer, is it true?
t
tall-airline-16756
02/18/2026, 12:23 PMYou're right to notice a difference — AgentServer was introduced in a more recent LiveKit Agents release, while older examples used the
WorkerOptionscli.run_appWorkerOptionscli.run_appAgentServerAgentServerload_fncload_thresholdWorkerOptionsAgentServerAgentServerWorkerOptionsAgentServerload_fncentrypointload_fncb
boundless-author-7811
02/23/2026, 7:41 AMtell me how does the AgentServer works and can be used to calculate load function if i am running livekit agent on 4 pods. How can i make sure the least busy pod agent picks the jobs and continue?
t
tall-airline-16756
02/23/2026, 7:42 AMThanks — I’ll explain how AgentServer works across multiple pods, how
load_fncWorkerOptionsload_fncload_thresholdload_fncload_thresholdload_fncload_thresholdload_fncload_thresholdload_thresholdcli.run_app(WorkerOptions(...))load_fnc Copy code
from livekit.agents import AgentServer
def compute_load(agent_server: AgentServer) -> float:
# Example: normalize active job count to [0,1] (assume 10 is "full")
max_jobs = 10
return min(len(agent_server.active_jobs) / max_jobs, 1.0)
server = AgentServer(
agent_name="outbound-caller",
port=8001,
prometheus_port=int(os.getenv("PROMETHEUS_PORT", "9090")),
load_threshold=0.9, # pod stops accepting new jobs above this
)
server.load_fnc = compute_load
# Start the AgentServer (this will register the pod with LiveKit)
server.run()load_fncentrypointprewarmload_fncload_fncb
boundless-author-7811
02/23/2026, 7:44 AMi will be using AgentServer, which is the optimum use case to follow?
t
tall-airline-16756
02/23/2026, 7:45 AMGreat — since you’ll be using AgentServer, here’s a focused, practical guide on the optimum way to run it across multiple Kubernetes pods so the least-busy pod picks up jobs reliably.
Acknowledgement
• You’ll run AgentServer in multiple pods (e.g., 4). Below I’ll cover how AgentServer works in that setup, how to implement a good
load_fncagent_nameload_fncload_thresholdload_fncload_threshold Copy code
def compute_load(server: AgentServer) -> float:
max_jobs = 10
return min(len(server.active_jobs) / max_jobs, 1.0) Copy code
# cpu_percent and mem_percent are 0..100
load = 0.6 * (cpu_percent / 100) + 0.3 * (mem_percent / 100) + 0.1 * (active_jobs / max_jobs)
load = max(0.0, min(load, 1.0))agent_nameload_thresholdload_thresholdterminationGracePeriodSecondsload_fnccli.run_app(WorkerOptions(...))load_fncentrypointprewarmload_fncb
boundless-author-7811
02/23/2026, 8:49 AMof the above 2 solutions, which is the more optimal one?
t
tall-airline-16756
02/23/2026, 8:51 AMThanks — great question. Between the two broad approaches we discussed (1) migrating to AgentServer with a custom
load_fncload_fncload_fncWorkerOptionsAgentServerentrypointprewarmload_fncload_thresholdload_fncb
boundless-author-7811
02/23/2026, 8:57 AMcan the load_fnc be custom?
t
tall-airline-16756
02/23/2026, 8:57 AMYes —
load_fncload_fncload_fncload_threshold Copy code
def compute_load(server: AgentServer) -> float:
max_jobs = 10
return min(len(server.active_jobs) / max_jobs, 1.0) Copy code
def compute_load(server: AgentServer) -> float:
cpu_percent = read_cpu_percent() # returns 0..100
active_jobs = len(server.active_jobs)
max_jobs = 10
load = 0.6 * (cpu_percent / 100.0) + 0.4 * (active_jobs / max_jobs)
return max(0.0, min(load, 1.0)) Copy code
def compute_load(server: AgentServer) -> float:
mem_used, mem_total = read_memory()
mem_fraction = mem_used / mem_total
return min(mem_fraction, 1.0)max_jobsload_thresholdload_thresholdload_thresholdload_fncAgentServerload_fncb
boundless-author-7811
02/23/2026, 9:04 AMi want the load function to be calculated on the basis of the below:
• i have 100000 tokens/minute available at the LLM ends, each successful call will definitely utilised some token. so every passing minute the available tokens should be calculated.
• i will have n number of phone numbers available for dispatching the calls, with each number capable of handling 50 calls each
based on the above information, how should we proceed?
t
tall-airline-16756
02/23/2026, 9:05 AMThanks — this is a great, concrete set of constraints. I’ll show a robust, production-ready approach you can drop into AgentServer so your
load_fncload_fncload_threshold Copy code
import time
from collections import deque
import threading
from typing import Deque, Tuple
# 1) Token rate tracker (per pod)
class TokenRateTracker:
def __init__(self, max_tokens_per_minute: int = 100000, window_seconds: int = 60):
self.max_tokens_per_sec = max_tokens_per_minute / 60.0
self.window_seconds = window_seconds
self.history: Deque[Tuple[float, int]] = deque() # (timestamp, token_count)
self.lock = threading.Lock()
def add_tokens(self, tokens: int):
now = time.time()
with self.lock:
self.history.append((now, tokens))
self._evict_old(now)
def _evict_old(self, now: float):
cutoff = now - self.window_seconds
while self.history and self.history[0][0] < cutoff:
self.history.popleft()
def get_token_load(self) -> float:
now = time.time()
with self.lock:
self._evict_old(now)
total_tokens = sum(t for _, t in self.history)
avg_per_sec = total_tokens / max(self.window_seconds, 1)
load = avg_per_sec / max(self.max_tokens_per_sec, 1e-9)
return min(max(load, 0.0), 1.0)
# 2) Phone capacity tracker (per pod)
class PhoneCapacityTracker:
def __init__(self, num_numbers: int, per_number_capacity: int = 50):
self.max_calls = max(1, num_numbers) * per_number_capacity
self.current_calls = 0
self.lock = threading.Lock()
def incr(self) -> int:
with self.lock:
self.current_calls += 1
return self.current_calls
def decr(self) -> int:
with self.lock:
self.current_calls = max(0, self.current_calls - 1)
return self.current_calls
def get_phone_load(self) -> float:
with self.lock:
load = self.current_calls / max(self.max_calls, 1)
return min(max(load, 0.0), 1.0)prewarm Copy code
TOKEN_TRACKER = TokenRateTracker(max_tokens_per_minute=100000, window_seconds=60)
PHONE_TRACKER = PhoneCapacityTracker(num_numbers=NUM_NUMBERS, per_number_capacity=50) Copy code
# after a successful LLM call where tokens_used is known:
TOKEN_TRACKER.add_tokens(tokens_used) Copy code
PHONE_TRACKER.incr()
# ... on job finish
PHONE_TRACKER.decr()load_fnc Copy code
def custom_load_fn(agent_server) -> float:
token_load = TOKEN_TRACKER.get_token_load()
phone_load = PHONE_TRACKER.get_phone_load()
# Optionally add smoothing, hysteresis, or small safety margin
load = max(token_load, phone_load)
return min(max(load, 0.0), 1.0) Copy code
server = AgentServer(
agent_name="outbound-caller",
load_threshold=0.9,
# other args...
)
server.load_fnc = custom_load_fn
server.run()load_thresholdentrypointprewarmcli.run_appb
boundless-author-7811
02/23/2026, 9:09 AMif the constraint are a global occurrence will that be a better approach? In redis we can the global capacity set and each pod can reference the same set of keys when calculation the load threshold?
t
tall-airline-16756
02/23/2026, 9:09 AMYes — if your LLM token budget and phone-number capacity are global across all pods, using Redis (or another shared store) is the better approach. It ensures every pod calculates load from the same authoritative state and prevents oversubscription that per-pod local tracking could allow.
Below is a concise plan, key correctness points, and a practical Redis-based pattern you can use for a global
load_fncload_fncload_fnc Copy code
-- args: KEYS[1]=token_key, ARGV[1]=tokens_to_add, ARGV[2]=max_tokens, ARGV[3]=window_ttl_seconds
local current = tonumber(redis.call("GET", KEYS[1]) or "0")
local add = tonumber(ARGV[1])
local max = tonumber(ARGV[2])
if current + add > max then
return {0, current} -- cannot reserve
end
current = redis.call("INCRBY", KEYS[1], add)
redis.call("EXPIRE", KEYS[1], tonumber(ARGV[3]))
return {1, current} -- reserved, return new total Copy code
import redis
from typing import Tuple
r = redis.Redis(...)
# keys/config
TOKENS_KEY = "global:tokens:current" # bucket for current minute (or finer buckets)
TOKENS_MAX = 100000
TOKENS_WINDOW_SECS = 60
PHONES_ACTIVE_KEY = "global:phones:active"
PHONES_MAX = NUM_NUMBERS * 50
def try_reserve_tokens(tokens_needed: int) -> bool:
# Call the Lua script (load it once and reuse)
res = r.evalsha(lua_sha, 1, TOKENS_KEY, tokens_needed, TOKENS_MAX, TOKENS_WINDOW_SECS)
return res[0] == 1
def increment_call_counter() -> int:
return r.incr(PHONES_ACTIVE_KEY)
def decrement_call_counter() -> int:
val = r.decr(PHONES_ACTIVE_KEY)
if val < 0:
r.set(PHONES_ACTIVE_KEY, 0)
return 0
return val
def get_global_load() -> float:
used_tokens = int(r.get(TOKENS_KEY) or 0)
phone_active = int(r.get(PHONES_ACTIVE_KEY) or 0)
token_load = min(used_tokens / TOKENS_MAX, 1.0)
phone_load = min(phone_active / PHONES_MAX, 1.0)
return max(token_load, phone_load)load_fncload_fncload_fncb
boundless-author-7811
02/23/2026, 9:29 AMi am using a scheduler to dispatch the job, the livekit server will pick the job and assign to the pods? or do i need to do it based on some logic?
t
tall-airline-16756
02/23/2026, 9:29 AMGood question — here’s how dispatch works and what your scheduler needs to do.
Short answer
• You do not need to implement pod-selection logic. The LiveKit server handles dispatch: when your scheduler creates a job (room/call) via LiveKit, LiveKit picks one of the registered AgentServer pods (same
agent_nameload_fncload_thresholdload_fncload_thresholdagent_nameload_fncload_thresholdload_fncload_fncload_fncload_fncb
boundless-author-7811
02/23/2026, 9:31 AMas of now, my scheduler does the below:
Copy code
self.lkapi.agent_dispatch.create_dispatch(
CreateAgentDispatchRequest(
agent_name=self.agent_name,
room=room_name,
metadata=json.dumps({
"trunk_id": trunk_id,
"phone_number": phone_number,
"customer_data": customer_row,
"caller_phone_id": phone_number_id,
"resource_type": "phone"
})
)
),
timeout=10.0
)t
tall-airline-16756
02/23/2026, 9:31 AMThanks — this is exactly the place to add global reservation logic. I’ll confirm how your current call behaves, explain where to add Redis reservations, and give a concrete Python pattern you can plug into your scheduler plus the matching AgentServer
load_fncagent_dispatch.create_dispatch(...)agent_namecreate_dispatchload_fnccreate_dispatchcreate_dispatchcreate_dispatchcreate_dispatch Copy code
-- token_reserve.lua
-- KEYS[1] = token_bucket_key
-- ARGV[1] = tokens_to_add
-- ARGV[2] = max_tokens_in_window
-- ARGV[3] = window_ttl_secs
local current = tonumber(redis.call("GET", KEYS[1]) or "0")
local add = tonumber(ARGV[1])
local max = tonumber(ARGV[2])
if current + add > max then
return {0, current}
end
local new = redis.call("INCRBY", KEYS[1], add)
redis.call("EXPIRE", KEYS[1], tonumber(ARGV[3]))
return {1, new} Copy code
import redis
import json
import time
from livekit import CreateAgentDispatchRequest # pseudo-import
r = redis.Redis(host="redis.svc", port=6379, db=0)
# keys / config
TOKENS_KEY = "global:tokens:current" # bucket covering the sliding window
TOKENS_MAX = 100_000
TOKENS_WINDOW_SECS = 60
PHONES_ACTIVE_KEY = "global:phones:active"
PHONES_MAX = NUM_NUMBERS * 50
# load the lua script once
with open("token_reserve.lua", "r") as f:
lua = f.read()
lua_sha = r.script_load(lua)
def try_reserve_tokens(tokens_needed: int) -> bool:
res = r.evalsha(lua_sha, 1, TOKENS_KEY, tokens_needed, TOKENS_MAX, TOKENS_WINDOW_SECS)
# res -> [status (0/1), new_total]
return bool(res[0])
def try_reserve_phone_slot() -> bool:
new_val = r.incr(PHONES_ACTIVE_KEY)
if new_val > PHONES_MAX:
# roll back
r.decr(PHONES_ACTIVE_KEY)
return False
return True
def release_phone_slot():
val = r.decr(PHONES_ACTIVE_KEY)
if val < 0:
r.set(PHONES_ACTIVE_KEY, 0)
def schedule_job(room_name, trunk_id, phone_number, phone_number_id, customer_row, tokens_estimate):
# 1) Reserve phone slot
if not try_reserve_phone_slot():
return {"ok": False, "reason": "no_phone_capacity"}
# 2) Reserve tokens atomically
if not try_reserve_tokens(tokens_estimate):
# rollback phone slot
release_phone_slot()
return {"ok": False, "reason": "no_token_capacity"}
# 3) Create LiveKit dispatch
try:
resp = self.lkapi.agent_dispatch.create_dispatch(
CreateAgentDispatchRequest(
agent_name=self.agent_name,
room=room_name,
metadata=json.dumps({
"trunk_id": trunk_id,
"phone_number": phone_number,
"customer_data": customer_row,
"caller_phone_id": phone_number_id,
"resource_type": "phone"
})
),
timeout=10.0
)
except Exception as e:
# Rollback both reservations on failure
release_phone_slot()
# Optionally reduce token bucket: because tokens were reserved by INCRBY in Lua,
# you should compensating-decrement the token key. Use a safe Lua script or DECRBY.
r.decrby(TOKENS_KEY, tokens_estimate)
return {"ok": False, "reason": "create_dispatch_failed", "error": str(e)}
# on success, return dispatch info
return {"ok": True, "dispatch": resp}load_fncload_fnc Copy code
def agent_load_fnc(agent_server) -> float:
try:
used_tokens = int(r.get(TOKENS_KEY) or 0)
phone_active = int(r.get(PHONES_ACTIVE_KEY) or 0)
token_load = min(used_tokens / TOKENS_MAX, 1.0)
phone_load = min(phone_active / PHONES_MAX, 1.0)
return max(token_load, phone_load)
except Exception:
# conservative fallback when Redis is unavailable
return 1.0server.load_fnc = agent_load_fncb
boundless-author-7811
02/23/2026, 12:52 PMhow can we know in the scheduler if the job that was dispatched was picked by the agent worker or was rejected, is there a way to know?
t
tall-airline-16756
02/23/2026, 12:52 PMGood question — you’re right to want a clear way for the scheduler to know whether a dispatched job was actually picked up by an agent or rejected. Below is a concise explanation of how dispatch works, what the scheduler call confirms, and practical patterns you can use to observe acceptance or rejection.
What the scheduler call confirms
• Calling agent_dispatch.create_dispatch schedules the dispatch with LiveKit, but it does NOT synchronously confirm the job was accepted by a worker. It only confirms the dispatch was created in LiveKit.
How to know whether a job was picked up or rejected
• Agent acceptance/rejection is asynchronous. An agent worker explicitly accepts or rejects the JobRequest (e.g.,
await job_request.accept(...)await job_request.reject()b
boundless-author-7811
02/23/2026, 1:00 PMbut livekit server must be getting some feedback about the job acceptance or rejection? Can we not use that as the feedback?
t
tall-airline-16756
02/23/2026, 1:00 PMShort answer: LiveKit does know when an agent accepts or rejects a JobRequest, but that acceptance is not returned synchronously from the
create_dispatchjob_request.accept(...)job_request.reject()create_dispatchcreate_dispatchcreate_dispatchdispatch_idb
boundless-author-7811
02/23/2026, 1:14 PMwhich is the most optimal solution, provide example of that
t
tall-airline-16756
02/23/2026, 1:15 PMThanks — good catch. I’ve completed the truncated portion and added the final notes you requested. Below is the finished agent_load_fnc and a short conclusion with tuning and reliability tips.
Complete AgentServer load_fnc (reads Redis keys)
Copy code
# agent_load_fn.py
def agent_load_fnc(agent_server) -> float:
try:
used_tokens = int(r.get(TOKEN_KEY) or 0)
active_phones = int(r.get(PHONES_KEY) or 0)
token_load = min(used_tokens / TOKENS_MAX, 1.0)
phone_load = min(active_phones / PHONES_MAX, 1.0)
load = max(token_load, phone_load)
# clamp to [0.0, 1.0] and return
return max(0.0, min(load, 1.0))
except Exception:
# conservative fallback if Redis is unavailable or an error occurs
return 1.0load_thresholdentrypointprewarmb
boundless-author-7811
02/23/2026, 1:16 PMIf you want, I can provide:
• A webhook-based example (scheduler side) that listens for participant_joined and maps to dispatch_id.
• An agent-side snippet that calls your scheduler endpoint immediately on accept/reject.
• A complete flow tying your Redis reservation + create_dispatch + accept/reject handling together.
which is the most optimal solution among the above, give example of that approach
t
tall-airline-16756
02/23/2026, 1:16 PMThe most optimal solution is a hybrid: use LiveKit webhooks (participant_joined) as your authoritative, low-latency signal that a dispatched job was picked up, and augment that with an agent-side immediate callback for the fastest confirmation and easier correlation. This gives you reliability (webhook = truth of agent in-room) and speed (agent callback = near-instant confirmation). Use a unique dispatch_id in dispatch metadata so both signals correlate to the scheduler record.
Why this hybrid is best
• Webhooks (participant_joined) are reliable and reflect the actual outcome (agent physically joined the room). They work even if the agent or scheduler restarts.
• Agent-side callbacks let your scheduler react faster (no need to wait for a webhook or poll the API).
• Combining both avoids false positives/negatives and simplifies recovery: if the callback is missed, the webhook still confirms acceptance; if webhook delays, the callback provides immediate feedback.
• Works well with global reservations (Redis): reserve before dispatch; release on callback or webhook/finalization.
What you need
• Include dispatch_id in the CreateAgentDispatchRequest metadata.
• Scheduler exposes an endpoint to receive agent callbacks.
• Configure LiveKit webhooks (participant_joined) to send events to your scheduler.
• Reservation logic (Redis) runs before dispatch; rollback on failure or on reject.
Example: Scheduler + webhook + agent callback (minimal)
1) Scheduler: create dispatch with dispatch_id
Copy code
import uuid, json
dispatch_id = str(uuid.uuid4())
resp = self.lkapi.agent_dispatch.create_dispatch(
CreateAgentDispatchRequest(
agent_name=self.agent_name,
room=room_name,
metadata=json.dumps({
"dispatch_id": dispatch_id,
"trunk_id": trunk_id,
"phone_number": phone_number,
"resource_type": "phone"
})
),
timeout=10.0
)
# Start a timeout watcher (see below) to handle no-accept cases. Copy code
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/dispatch-callback", methods=["POST"])
def dispatch_callback():
data = request.json # { dispatch_id, status: "accepted"|"rejected", agent_id, reason? }
dispatch_id = data["dispatch_id"]
status = data["status"]
# Lookup your scheduled job by dispatch_id, mark accepted/rejected, release/confirm reservations
if status == "accepted":
mark_dispatch_accepted(dispatch_id, data)
else:
mark_dispatch_rejected(dispatch_id, data)
return jsonify({"ok": True})dispatch_id Copy code
@app.route("/livekit-webhook", methods=["POST"])
def livekit_webhook():
ev = request.json
if ev.get("event") == "participant_joined":
participant = ev["participant"]
# Look in participant.metadata or room.metadata for dispatch_id (embed dispatch_id)
dispatch_id = find_dispatch_id_from_event(ev)
if dispatch_id:
mark_dispatch_accepted(dispatch_id, {"agent_joined": True})
return jsonify({"ok": True}) Copy code
import requests
def on_job_request(job_request):
dispatch_id = job_request.metadata.get("dispatch_id")
# decide whether to accept
accepted = should_accept_job(job_request)
if accepted:
job_request.accept()
# immediately notify scheduler
<http://requests.post|requests.post>("<https://scheduler.example/dispatch-callback>", json={
"dispatch_id": dispatch_id,
"status": "accepted",
"agent_id": my_agent_id,
}, timeout=1.0)
else:
job_request.reject()
<http://requests.post|requests.post>("<https://scheduler.example/dispatch-callback>", json={
"dispatch_id": dispatch_id,
"status": "rejected",
"reason": "busy_or_quota"
}, timeout=1.0)b
boundless-author-7811
02/24/2026, 9:37 AMcurrently i am running the same Agent in 2 pods with the below pod configuration:
Copy code
resources:
limits:
memory: 4Gi
requests:
cpu: 500m
memory: 1Git
tall-airline-16756
02/24/2026, 9:37 AMGreat — the hybrid (webhook + agent callback) approach you chose is the most optimal. It gives both fast confirmation (agent callback) and a reliable authoritative signal (LiveKit webhook when the agent actually joins). Below I’ll give a concise, copy-pasteable end-to-end example you can drop into your system: scheduler reservation + create_dispatch, agent callback on accept/reject, and webhook reconciliation (participant_joined). I’ll also include notes on timeouts, rollback, and safety.
Assumptions
• You use Redis for global reservations (tokens and phone slots).
• You include a unique
dispatch_id Copy code
import uuid, json, time, requests, redis
from livekit import CreateAgentDispatchRequest # pseudo-import for your SDK
r = redis.Redis(host="redis.svc", port=6379, db=0)
TOKENS_KEY = "global:tokens:current"
TOKENS_MAX = 100_000
TOKENS_WINDOW_SECS = 60
PHONES_KEY = "global:phones:active"
PHONES_MAX = NUM_NUMBERS * 50
# load token_reserve.lua once and store sha
with open("token_reserve.lua") as f:
token_lua = f.read()
token_sha = r.script_load(token_lua)
def try_reserve_tokens(tokens_needed: int) -> bool:
res = r.evalsha(token_sha, 1, TOKENS_KEY, tokens_needed, TOKENS_MAX, TOKENS_WINDOW_SECS)
return bool(res[0])
def try_reserve_phone_slot() -> bool:
new = r.incr(PHONES_KEY)
if new > PHONES_MAX:
r.decr(PHONES_KEY)
return False
return True
def release_phone_slot():
val = r.decr(PHONES_KEY)
if val < 0:
r.set(PHONES_KEY, 0)
def release_tokens(tokens: int):
# Use DECRBY safely; if key missing, set to 0
r.decrby(TOKENS_KEY, tokens)
if int(r.get(TOKENS_KEY) or 0) < 0:
r.set(TOKENS_KEY, 0)
def schedule_call(room_name, trunk_id, phone_number, phone_number_id, customer_row, tokens_estimate):
dispatch_id = str(uuid.uuid4())
# 1) Reserve phone slot
if not try_reserve_phone_slot():
return {"ok": False, "reason": "no_phone_capacity"}
# 2) Reserve tokens
if not try_reserve_tokens(tokens_estimate):
release_phone_slot()
return {"ok": False, "reason": "no_token_capacity"}
# 3) Create dispatch
try:
resp = self.lkapi.agent_dispatch.create_dispatch(
CreateAgentDispatchRequest(
agent_name=self.agent_name,
room=room_name,
metadata=json.dumps({
"dispatch_id": dispatch_id,
"trunk_id": trunk_id,
"phone_number": phone_number,
"caller_phone_id": phone_number_id,
"resource_type": "phone"
})
),
timeout=10.0
)
except Exception as e:
# rollback reservations
release_phone_slot()
release_tokens(tokens_estimate)
return {"ok": False, "reason": "create_dispatch_failed", "error": str(e)}
# 4) Start a watcher (background) to reconcile if no accept in time
start_accept_watcher(dispatch_id, tokens_estimate, timeout=20) # implement per your infra
return {"ok": True, "dispatch_id": dispatch_id, "dispatch": resp}token_reserve.luastart_accept_watcher Copy code
import requests
SCHEDULER_CALLBACK_URL = "<https://scheduler.example/dispatch-callback>"
SCHEDULER_TIMEOUT = 1.0
def on_job_request(job_request):
metadata = job_request.metadata or {}
dispatch_id = metadata.get("dispatch_id")
if should_accept_job(job_request):
# Accept the job with LiveKit
job_request.accept()
# Best-effort notify scheduler (non-blocking ideally)
try:
<http://requests.post|requests.post>(SCHEDULER_CALLBACK_URL, json={
"dispatch_id": dispatch_id,
"status": "accepted",
"agent_id": MY_AGENT_ID,
}, timeout=SCHEDULER_TIMEOUT)
except Exception:
# ignore — webhook will still confirm when participant joins
pass
else:
job_request.reject()
try:
<http://requests.post|requests.post>(SCHEDULER_CALLBACK_URL, json={
"dispatch_id": dispatch_id,
"status": "rejected",
"reason": "quota_or_busy",
"agent_id": MY_AGENT_ID,
}, timeout=SCHEDULER_TIMEOUT)
except Exception:
pass Copy code
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/dispatch-callback", methods=["POST"])
def dispatch_callback():
data = request.json
dispatch_id = data["dispatch_id"]
status = data.get("status")
if status == "accepted":
mark_dispatch_accepted(dispatch_id, data)
elif status == "rejected":
mark_dispatch_rejected(dispatch_id, data)
# release reservations
release_phone_slot_for_dispatch(dispatch_id)
release_tokens_for_dispatch(dispatch_id)
return jsonify({"ok": True})
@app.route("/livekit-webhook", methods=["POST"])
def livekit_webhook():
ev = request.json
if ev.get("event") == "participant_joined":
# Extract dispatch_id from room or participant metadata
dispatch_id = extract_dispatch_id(ev)
if dispatch_id:
mark_dispatch_accepted(dispatch_id, {"source": "webhook", "event": ev})
# Keep reservations confirmed; no action needed
return jsonify({"ok": True}) Copy code
def start_accept_watcher(dispatch_id, tokens_reserved, timeout=20):
# schedule background task to run after timeout
def watcher():
time.sleep(timeout)
if not is_dispatch_accepted(dispatch_id):
# release reservations
release_phone_slot_for_dispatch(dispatch_id)
release_tokens(tokens_reserved)
mark_dispatch_timed_out(dispatch_id)
spawn_background_thread(watcher)dispatch_id2 Views