Overview
The AgentLauncher handles warmup and lifecycle management for agents. It ensures all components (LLM, TTS, STT, turn detection) are warmed up before agents are launched and manages multiple concurrent sessions.
from vision_agents.core import Agent, AgentLauncher
def create_agent():
return Agent(
edge=Edge(),
agent_user=User(id="agent", name="AI Assistant"),
llm=Realtime(),
instructions="You are a helpful assistant.",
)
async def join_call(agent, call_type, call_id):
call = await agent.create_call(call_type, call_id)
async with agent.join(call):
await agent.finish()
launcher = AgentLauncher(
create_agent=create_agent,
join_call=join_call,
agent_idle_timeout=60.0,
max_concurrent_sessions=10,
)
async with launcher:
session = await launcher.start_session("my-call-id")
await session.wait()
Constructor
AgentLauncher(
create_agent: Callable[..., Agent | Coroutine[Any, Any, Agent]],
join_call: Callable[[Agent, str, str], Coroutine],
agent_idle_timeout: float = 60.0,
max_concurrent_sessions: Optional[int] = None,
max_sessions_per_call: Optional[int] = None,
max_session_duration_seconds: Optional[float] = None,
maintenance_interval: float = 5.0,
registry: Optional[SessionRegistry] = None,
)
Parameters
create_agent
Callable[..., Agent | Coroutine[Any, Any, Agent]]
required
A function that creates and returns an Agent instance. Can be sync or async.
join_call
Callable[[Agent, str, str], Coroutine]
required
A coroutine function that handles joining a call with the agent. Receives (agent, call_type, call_id).
Timeout in seconds for an agent to stay alone on a call before being automatically closed. Set to 0 to disable idle timeout (agents won’t leave until the call ends).
max_concurrent_sessions
Optional[int]
default:"None"
Maximum number of concurrent sessions allowed across all calls. None means unlimited.
max_sessions_per_call
Optional[int]
default:"None"
Maximum number of sessions allowed per call_id. None means unlimited.
max_session_duration_seconds
Optional[float]
default:"None"
Maximum duration in seconds for a session before it is automatically closed. None means unlimited.
Interval in seconds between cleanup checks for idle or expired sessions.
registry
Optional[SessionRegistry]
default:"None"
Optional SessionRegistry for multi-node session management. When provided, sessions are registered in shared storage and heartbeats are sent on every cleanup interval.
Properties
warmed_up
@property
def warmed_up(self) -> bool
Returns True if the agent components have been warmed up.
running
@property
def running(self) -> bool
Returns True if the launcher is currently running.
ready
@property
def ready(self) -> bool
Returns True if the launcher is warmed up and running.
registry
@property
def registry(self) -> SessionRegistry
Returns the session registry instance.
Methods
start
async def start(self) -> None
RuntimeError: If the launcher is already running.
stop
async def stop(self) -> None
warmup
async def warmup(self) -> None
launch
async def launch(self, **kwargs) -> Agent
Additional keyword arguments to pass to create_agent.
start_session
async def start_session(
self,
call_id: str,
call_type: str = "default",
video_track_override_path: Optional[str] = None,
) -> AgentSession
Unique identifier for the call to join. Must contain only a-z, 0-9, _ and -.
video_track_override_path
Optional[str]
default:"None"
Optional path to a video file to use instead of a live video track.
An AgentSession object representing the new session.
InvalidCallId: If the call_id contains invalid characters.MaxConcurrentSessionsExceeded: If the maximum number of concurrent sessions has been reached.MaxSessionsPerCallExceeded: If the maximum number of sessions for this call_id has been reached.
close_session
async def close_session(
self,
session_id: str,
wait: bool = False
) -> bool
When True, wait for the underlying agent to finish. Otherwise, just cancel the task and return.
True if session was found and closed, False otherwise.
get_session
def get_session(self, session_id: str) -> Optional[AgentSession]
The session ID to look up.
The AgentSession if found on this node, None otherwise.
request_close_session
async def request_close_session(
self,
call_id: str,
session_id: str
) -> None
The call the session belongs to.
get_session_info
async def get_session_info(
self,
call_id: str,
session_id: str
) -> SessionInfo | None
The call the session belongs to.
Session information if found, None otherwise.
AgentSession
Represents an active agent session within a call. An AgentSession wraps an Agent instance along with metadata about the session.
Properties
agent
The Agent instance for this session.
call_id
The call ID this session belongs to.
started_at
Timestamp when the session started.
task
The async task running the agent’s call handler.
finished
@property
def finished(self) -> bool
@property
def id(self) -> str
Methods
wait
Wait for the session task to finish running.
on_call_for
def on_call_for(self) -> float
idle_for
def idle_for(self) -> float
Context Manager Usage
The AgentLauncher supports async context manager usage:
async with launcher:
# Launcher is started automatically
session = await launcher.start_session("call-123")
await session.wait()
# Launcher is stopped automatically