Infrastructure | SignalWire

These classes mirror the LiveKit infrastructure types used in agent entrypoints. On SignalWire, connection lifecycle and room management are handled automatically by the control plane, so these are lightweight stubs that maintain API compatibility.

JobContext

Mirrors a LiveKit JobContext. Provides access to the room and process objects, and lifecycle methods that the entrypoint calls before starting a session.

1 from signalwire.livewire import JobContext
2 
3 async def entrypoint(ctx: JobContext):
4     await ctx.connect()
5     await ctx.wait_for_participant()
6     # ctx.room and ctx.proc are available

Properties

room

Room

A Room stub instance.

proc

JobProcess

A JobProcess instance with a userdata dict, populated by the setup function if one was registered on AgentServer.

Methods

connect

connect() -> None

No-op. SignalWire’s control plane handles connection lifecycle automatically. The agent connects when the platform invokes the SWML endpoint.

wait_for_participant

wait_for_participant(identity=None) -> None

No-op. SignalWire handles participant management automatically.

identity

Optional[str]Defaults to None

Participant identity to wait for. Accepted for API compatibility.

JobProcess

Mirrors a LiveKit JobProcess. Used for prewarm and setup tasks. The setup function registered on AgentServer.setup_fnc receives a JobProcess instance.

1 from signalwire.livewire import JobProcess
2 
3 def on_setup(proc: JobProcess):
4     proc.userdata["db_connection"] = connect_to_db()

Properties

userdata

dict[str, Any]

Arbitrary data dict for sharing state between the setup function and the entrypoint. Defaults to an empty dict.

Room

A stub representing the LiveKit room abstraction. On SignalWire, rooms are managed by the control plane and this object is a placeholder for API compatibility.

Properties

name

strDefaults to "livewire-room"

The room name. Always "livewire-room" in the current implementation.

Example

1 from signalwire.livewire import (
2     Agent,
3     AgentSession,
4     AgentServer,
5     JobContext,
6     JobProcess,
7     run_app,
8 )
9 
10 server = AgentServer()
11 
12 def setup(proc: JobProcess):
13     # Pre-warm: load config, connect to databases, etc.
14     proc.userdata["greeting"] = "Welcome to Acme Corp!"
15 
16 server.setup_fnc = setup
17 
18 @server.rtc_session()
19 async def entrypoint(ctx: JobContext):
20     await ctx.connect()
21     await ctx.wait_for_participant()
22 
23     greeting = ctx.proc.userdata.get("greeting", "Hello!")
24 
25     agent = Agent(instructions="You are a helpful assistant.")
26     session = AgentSession()
27     await session.start(agent, room=ctx.room)
28     session.say(greeting)
29 
30 run_app(server)