https://livekit.io logo
Join Slack
Powered by
What is the ideal way to run a callback function w...
# ask-ai
e
What is the ideal way to run a callback function when the agent session is started [actually connected] and when the session ends.
t
Thanks for reaching out, Aditya. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m responding to your question about when to run callbacks for an agent session: specifically once the session is actually connected, and when it ends. Short answer • Run your “session started” callback immediately after you call 
await session.start()
or 
await ctx.connect()
inside the 
@server.rtc_session
entrypoint — that guarantees the agent is connected before the callback runs. • Register an “on session end” callback with 
ctx.add_shutdown_callback(...)
or provide an 
on_session_end
function to 
@server.rtc_session(...)
for cleanup/reporting when the session closes. Why and how (concise examples) 1) Session started (after actual connection) • Place your callback logic after the connect/start call in the entrypoint so it runs only once the connection is established. Example:
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    # start/connect the agent/session
    await session.start(agent=MyAgent(), room=ctx.room)
    # or: await ctx.connect()

    # Now the agent is actually connected — run your callback
    await on_session_started(ctx)
2) Session end (cleanup / reporting) • Use 
ctx.add_shutdown_callback()
to register async cleanup or notification logic that runs when the session shuts down. Example:
Copy code
async def on_shutdown():
    <http://logger.info|logger.info>("Session ended — cleanup running")
    await do_cleanup_tasks()

ctx.add_shutdown_callback(on_shutdown)
• Alternatively, for a higher-level hook, pass an 
on_session_end
callable to the 
@server.rtc_session
decorator to receive the session end report: Example:
Copy code
def on_session_end(report: SessionEndReport):
    <http://logger.info|logger.info>(f"Session ended: {report}")

@server.rtc_session(agent_name="my-agent", on_session_end=on_session_end)
async def entrypoint(ctx: JobContext):
    ...
Notes and tips • Choose 
ctx.add_shutdown_callback(...)
when you need per-session async cleanup that runs as part of the session lifecycle. • Use the decorator 
on_session_end
when you want a simple sync-style callback that receives the session end report for logging or analytics. • Keep the “session started” callback lightweight if it runs immediately after connect (e.g., kick off background tasks rather than blocking long work). If you want, tell me which pattern you’re using now (manual 
ctx.connect()
vs 
session.start()
vs another flow) and I’ll tailor a small code example for your exact setup. Sources: Job lifecycle | LiveKit Documentation | Agent session | LiveKit Documentation
e
I am using session.start
t
Thanks — since you’re using 
session.start()
, here’s a concise, practical pattern you can use to run a callback after the agent is actually connected and to run cleanup when the session ends. 1) Run a “session started” callback right after 
await session.start(...)
• That guarantees the agent is connected before your callback runs. • Keep any heavy work off the main start path (spawn background tasks if needed). Example:
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    # Connect and start the agent session
    await session.start(agent=MyAgent(), room=ctx.room)

    # Agent is now connected — run your start callback
    await on_session_started(ctx)
2) Register an “on session end” shutdown callback with 
ctx.add_shutdown_callback(...)
• This runs when the session closes and is ideal for async cleanup, logging, or notifying other services. Example:
Copy code
async def on_shutdown():
    <http://logger.info|logger.info>("Session ended — running cleanup")
    await cleanup_resources()

# Register the shutdown callback early in the entrypoint
ctx.add_shutdown_callback(on_shutdown)
Recommended full pattern (register shutdown callback first, then start the session):
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    # Register shutdown callback before starting so it always runs on exit
    ctx.add_shutdown_callback(on_shutdown)

    # Start/connect the session
    await session.start(agent=MyAgent(), room=ctx.room)

    # Session is connected — run any on-start actions
    await on_session_started(ctx)

    # Your normal session work continues here...
Quick tips • Register 
ctx.add_shutdown_callback
before 
session.start()
so it’s guaranteed to run even if start raises or the session terminates unexpectedly. • For long-running or blocking post-start actions, start background tasks (e.g., create an asyncio.Task) so the entrypoint can continue handling session lifecycle. • If you prefer a synchronous-style end hook that receives the final report, you can also pass 
on_session_end
to 
@server.rtc_session(...)
for higher-level reporting. Would you like a tailored example showing background task creation after start, or an example that logs a SessionEndReport in 
on_session_end
?
e
Copy code
# Session is connected — run any on-start actions
    await on_session_started(ctx)
print a normal message on sessions start
t
Thanks — since you’re using 
session.start()
, here’s a minimal example that prints a normal message when the session is actually connected, plus a shutdown callback for when the session ends.
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    # Register shutdown callback so it always runs on session end
    async def on_shutdown():
        print("Session ended — cleanup running")
        # await cleanup_tasks()

    ctx.add_shutdown_callback(on_shutdown)

    # Start/connect the session
    await session.start(agent=MyAgent(), room=ctx.room)

    # Agent is now connected — print a normal message
    print("Session started — agent is connected")

    # Optional: run additional on-start async logic
    # await on_session_started(ctx)

    # Continue with your session work...
Notes • Place the print immediately after 
await session.start(...)
— that guarantees the connection is established. • Register 
ctx.add_shutdown_callback(...)
before calling 
session.start()
so the end callback is always registered, even if start fails. • If your on-start work is long-running, start it as a background task with 
asyncio.create_task(...)
so it doesn't block the entrypoint. Would you like a version that logs with the 
logging
module instead of print, or one that starts a background task after start?
e
Copy code
# await on_session_started(ctx)
i dont understand this line, what is it ?
t
Good question — happy to clarify. The line 
await on_session_started(ctx)
is just a placeholder that calls a user-defined asynchronous function named 
on_session_started
, passing the 
ctx
(JobContext) to it. It means “run the async callback called 
on_session_started
and wait for it to finish.” Why it’s written that way • 
on_session_started
is intended to be a function you define that contains whatever logic you want to run right after the session is connected (logging, notifications, kicking off background work, etc.). • The 
await
keyword is used because the function is async (it returns a coroutine). Using 
await
runs the coroutine and pauses until it completes. How to define it (simple example)
Copy code
async def on_session_started(ctx: JobContext):
    # run whatever you want when the session is connected
    print("Session started — agent is connected")
    # you can access ctx.room, ctx.participants, etc.
    # await do_async_work()
How it fits into the entrypoint
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    ctx.add_shutdown_callback(on_shutdown)  # register end callback

    await session.start(agent=MyAgent(), room=ctx.room)

    # Call your on-start function and wait for it to finish
    await on_session_started(ctx)

    # continue with other session work...
If you don’t need to wait for the on-start work to finish, you can start it in the background instead:
Copy code
import asyncio
asyncio.create_task(on_session_started(ctx))  # fire-and-forget
Would you like a concrete on_start function that, for example, logs participant info or sends a notification? I can provide that.
e
raw_messages = session.history.to_dict( exclude_image=True, exclude_timestamp=False, exclude_audio=True, exclude_function_call=False, ).get("items", []) what does the response looks like including function tool ?
t
Thanks — you asked what the response looks like when you call:
Copy code
raw_messages = session.history.to_dict(
    exclude_image=True,
    exclude_timestamp=False,
    exclude_audio=True,
    exclude_function_call=False,
).get("items", [])
Short answer • You get a list of items under the "items" key. Each item is a dictionary with a 
type
field describing the kind of entry (e.g., 
user_message
, 
agent_message
, 
function_call
, 
function_call_output
, etc.). • When 
exclude_function_call=False
, function calls appear as 
function_call
items (with their arguments) and are followed by 
function_call_output
items (with the result and an error flag). Example (typical structure)
Copy code
[
  {
    "type": "user_message",
    "id": "msg_1",
    "timestamp": "2024-01-01T12:00:00Z",
    "content": "What's the weather in Tokyo?"
  },
  {
    "type": "function_call",
    "id": "call_1",
    "call_id": "fnk_12345",
    "name": "lookup_weather",
    "arguments": {
      "location": "Tokyo"
    }
  },
  {
    "type": "function_call_output",
    "id": "output_1",
    "call_id": "fnk_12345",
    "output": "sunny with a temperature of 70 degrees",
    "is_error": false,
    "timestamp": "2024-01-01T12:00:02Z"
  },
  {
    "type": "agent_message",
    "id": "msg_2",
    "timestamp": "2024-01-01T12:00:05Z",
    "content": "The weather in Tokyo is sunny with a temperature of 70 degrees."
  }
]
Key fields to expect • `type`: identifies the item type (
user_message
, 
agent_message
, 
function_call
, 
function_call_output
, etc.). • `id`: unique identifier for the item. • `timestamp`: present because you set 
exclude_timestamp=False
. • For `function_call`: ◦ `call_id`: ID used to correlate the call and its output. ◦ `name`: function/tool name. ◦ `arguments`: dictionary of the arguments passed. • For `function_call_output`: ◦ `call_id`: matches the `function_call`'s 
call_id
. ◦ `output`: the returned data (string, dict, or structured value). ◦ `is_error`: boolean indicating whether the function raised or returned an error. Notes and tips • You excluded images and audio, so those binary fields are not present in items. • If a function returns structured data (JSON/dict), 
output
may be that structure rather than a plain string. • Use the 
call_id
to correlate each 
function_call
with its 
function_call_output
. • If you need a concrete Python example to iterate and process these items (e.g., extract all function outputs), tell me and I’ll provide one. Sources: Agents and handoffs | LiveKit Documentation | Testing and evaluation | LiveKit Documentation | Building multi-agent architectures with LiveKit agents | LiveKit
2 Views
Open in Slack
PreviousNext
Powered by