Can you give me the code to add a ctx shutdown cal...
# ask-ai
b
Can you give me the code to add a ctx shutdown callback to print the entire transcript of the conversation? i am using pipeline mode, but the following code does not work async def print_transcript(): print("\n================ FULL CALL TRANSCRIPT ================\n") # Try primary history property try: history = getattr(session, "history", None) if history: for item in history: role = getattr(item, "role", None) or getattr(item, "speaker", "unknown") content = getattr(item, "content", None) or getattr(item, "text", "") print(f"{role.upper()}: {content}") else: print("session.history is empty or not available.") except Exception as e: print("Error reading session.history:", e) # Fallback to chat context if needed try: chat_ctx = getattr(session, "_chat_ctx", None) if chat_ctx: messages = getattr(chat_ctx, "messages", None) if messages: print("\n--- Fallback: _chat_ctx messages ---\n") for msg in messages: role = getattr(msg, "role", None) content = getattr(msg, "content", None) print(f"{str(role).upper()}: {content}") else: print("_chat_ctx exists but has no messages.") else: print("No _chat_ctx found.") except Exception as e: print("Error reading _chat_ctx:", e) print("\n================ END TRANSCRIPT =======================\n") ctx.add_shutdown_callback(print_transcript)
t
Thanks for reaching out, Aryan. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m responding about your request to add a ctx shutdown callback that prints the entire transcript when using pipeline mode, and the code you posted that “does not work.” Quick diagnosis • The most likely issue is that your callback cannot see the
session
object when it runs (scope/definition timing). The callback must be defined in the same scope where
session
is created so it can read
session.history
. Also make sure the callback is registered before shutdown and that shutdown is actually triggered. Working minimal example • This example shows the pattern that works in pipeline mode: create
session
inside the entrypoint, define the shutdown callback in that same scope so it can access
session.history
, register it with
ctx.add_shutdown_callback
, then start the session.
Copy code
@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    # create session in this scope
    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3-general"),
        llm=inference.LLM(model="openai/gpt-4o"),
        tts=inference.TTS(model="cartesia/sonic-3"),
        vad=silero.VAD.load(),
    )

    # shutdown callback defined in same scope so it can access `session`
    async def print_transcript():
        print("\n================ FULL CALL TRANSCRIPT ================\n")
        history = getattr(session, "history", None)
        if history:
            for item in history:
                role = getattr(item, "role", None) or getattr(item, "speaker", "unknown")
                content = getattr(item, "content", None) or getattr(item, "text", "")
                print(f"{role.upper()}: {content}")
        else:
            print("No conversation history available.")
        print("\n================ END TRANSCRIPT =======================\n")

    # register callback
    ctx.add_shutdown_callback(print_transcript)

    # start the session and connect
    await session.start(
        agent=Agent(instructions="You are a helpful assistant."),
        room=ctx.room,
    )
    await ctx.connect()
Notes and best practices • Use
session.history
(public API) to get the full conversation history. • Avoid relying on private/internal attributes like
_chat_ctx
. • The callback can be
async def
(safe even if not awaiting anything). If it needs to write files or call async APIs, keep it async and await those operations. • Register the callback before you call
session.start()
/
ctx.connect()
so it’s in place for shutdown. • Confirm shutdown is happening (add a simple print at top of callback); if you never see that print, the callback may not be invoked. If this still “does not work” Please share any of the following so I can narrow it down: 1. The exact behavior or error (traceback, logs, or “nothing prints”). 2. The entrypoint function (the surrounding code where you create session and register the callback). 3. Whether you see any prints from the callback at shutdown (to confirm it runs). 4. Whether you use
AgentSession
pipeline mode or a custom flow, and the framework version. If you provide one or two of those items I’ll give a tailored fix or point out the exact line that needs changing. Sources: Building multi-agent architectures with LiveKit agents | LiveKit | Transcriber | LiveKit Documentation | Server startup modes | LiveKit Documentation