This message was deleted.

Thanks for reaching out, Emily. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

yes

Great — below are concise examples showing how to access

session.llm

inside a tool call for both Python and TypeScript. Python (async tool):

from livekit import function_tool, RunContext

@function_tool()
async def lookup_weather(context: RunContext, location: str):
    chat_ctx = context.session.chat_ctx  # current conversation context
    async with context.session.llm.chat(chat_ctx=chat_ctx) as stream:
        response_text = ""
        async for chunk in stream:
            # stream chunks (tokens) as they arrive
            response_text += chunk.token
    return {"location": location, "weather": response_text}

TypeScript (async tool):

import { functionTool, RunContext } from "livekit";

export const lookupWeather = functionTool(async (context: RunContext, location: string) => {
  const chatCtx = context.session.chat_ctx;
  const stream = await context.session.llm.chat({ chat_ctx: chatCtx });
  let responseText = "";

  for await (const chunk of stream) {
    // process streaming chunks
    responseText += chunk.token;
  }

  return { location, weather: responseText };
});

Notes: • Use

context.session.chat_ctx

if you want the LLM to continue the current conversation context. • You can also make single-response calls if you prefer (non-streaming), depending on your LLM client API. • Tell me which pattern you want (streaming vs single response) or share your runtime and I’ll tailor the snippet.

would this work # Create a temporary chat context for this tool call temp_chat_ctx = ChatContext() temp_chat_ctx.messages.append( ChatMessage(role="user", content=prompt) ) # Use the session's LLM to generate a contextual spark response_text = "" async with run_context.session.llm.chat(chat_ctx=temp_chat_ctx) as stream: async for chunk in stream: response_text += chunk.choices[0].delta.content or ""

Yes — that approach will work. A few concise notes and a small tweak to ensure robustness: • Creating a temporary

ChatContext

and appending a

ChatMessage

with the prompt is fine; the LLM will use that context for the call. • Use the

run_context.session.llm.chat(chat_ctx=...)

call as you wrote to open a streaming chat. • In the stream loop, handle missing fields safely and prefer reading token text consistently. Suggested minor revision (preserves your logic, safer extraction of token content):

temp_chat_ctx = ChatContext()
temp_chat_ctx.messages.append(ChatMessage(role="user", content=prompt))

response_text = ""
async with run_context.session.llm.chat(chat_ctx=temp_chat_ctx) as stream:
    async for chunk in stream:
        # some providers put token text in different fields; this safely concatenates available content
        token = None
        try:
            token = chunk.choices[0].delta.content
        except Exception:
            token = getattr(chunk, "token", None)
        response_text += token or ""

If you want the temporary context to include prior conversation messages (so the LLM sees history plus the new prompt), I can show how to copy

run_context.session.chat_ctx

and append the new message instead. Which behavior do you want: fresh prompt-only context or history+prompt?

but it says 2025-12-22 161112,370 - INFO session-tools - Generating conversation spark for context: <livekit.agents.voice.events.RunContext object at 0x1158a16a0>... {"pid": 44817, "job_id": "AJ_BXN2qKv6WwhG"} 2025-12-22 161112,370 - ERROR session-tools - Error generating conversation spark: 'NoneType' object has no attribute 'chat' {"pid": 44817, "job_id": "AJ_BXN2qKv6WwhG"}

Thanks — that error means

run_context.session

(or

run_context.session.llm

) is None when you call

.chat

. Let’s narrow it down. Please check/send these quick diagnostics: 1. Where the tool is defined — paste the function signature and decorator you use (e.g.,

@function_tool()

and params). 2. How the tool is invoked — show the call path or the snippet that triggers the tool. 3. Add this runtime check right before the

.chat

call and share the printed results:

print("run_context:", run_context)
print("run_context.session:", getattr(run_context, "session", None))
print("run_context.session.llm:", getattr(getattr(run_context, "session", None), "llm", None))

Common causes and fixes: • The tool received a different context object (not RunContext). Ensure your tool signature accepts the correct context type and the agent runtime passes it. • Session was not initialized for that execution path. If

session

is None, ensure the agent/handler attaches a session before calling tools. • Variable naming mismatch (e.g., you used

context

elsewhere but

run_context

here). Use the same param name the framework provides. If you prefer, paste the tool function and the invocation snippet and I’ll point out the exact line to fix.

how come the run_context.session is fine but run_context.session.llm is none

Thanks — I’m addressing why

run_context.session

is valid but

run_context.session.llm

is None. Quick checklist and targeted diagnostics — please run or share the indicated items: 1) Confirm how you create the session • Paste the AgentSession (or session) initialization. It must include an

llm

(either a model id string or an LLM instance). Example:

session = AgentSession(llm="openai/gpt-4.1-mini", ...)

2) Confirm how you start the agent • Show the call that starts the agent or session (e.g.,

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

). If the agent is initialized outside the session flow, the LLM may not be attached. 3) Quick runtime print (right before your

.chat

call) • Add/return these prints and paste results:

print("session:", run_context.session)
print("session.llm:", getattr(run_context.session, "llm", None))
print("type(session.llm):", type(getattr(run_context.session, "llm", None)))

4) Possible causes to check now • You omitted

llm

when constructing the session (most common). • Your agent class or some initialization step overrides or sets

session.llm = None

. • You passed a plugin or different LLM wrapper that wasn’t assigned to

session.llm

(ensure you pass it into AgentSession). If you share the AgentSession construction and the agent start snippet, I’ll point out exactly where the LLM should be set or why it’s getting lost. Which code snippet can you paste? Sources: Large language models (LLM) | LiveKit Docs | Kimi LLM | LiveKit docs | LLM-Powered Content Filter | LiveKit docs | OpenAI LLM plugin guide | LiveKit docs

async def entrypoint(ctx: JobContext): # Initialize conversation logger conversation_logger = ConversationLogger(enabled=True) if TEST_MODE else None # Create user data and session (before connection to avoid audio buffer warning) userdata = UserData(session_config=None) session = AgentSession[UserData]( userdata=userdata, turn_detection="manual", ) # Create and start RoomIO before connecting room_io = RoomIO(session, room=ctx.room) await room_io.start() # Now connect to the room await ctx.connect() logger.info(f"Session started - Room: {ctx.room.name}") # Wait for participant to join logger.info("Waiting for participant to join...") participant = await ctx.wait_for_participant() logger.info(f"Participant joined: {participant.identity}") # Get session configuration if USE_DEFAULT_TEST_CONFIG: logger.info("Using default session config for test environment") session_config = get_default_session_config() else: session_config = await get_session_config(ctx.room, participant.identity) print("------------------ session_config -------------------") print(session_config) conversation_logger.log_message("system", session_config) if conversation_logger else None print("------------------------------------------------------") if not session_config: logger.warning("Failed to retrieve session config, using defaults") session_config = get_default_session_config() # Update user data with session config userdata.session_config = session_config # Assemble prompt from session config assembled_instructions = assemble_prompt_from_config(session_config) logger.info("Assembled instructions from session config") if TEST_MODE: conversation_logger.log_message("system", assembled_instructions) # Create agent and start session agent = MyAgent(instructions=assembled_instructions, conversation_logger=conversation_logger) await session.start(agent=agent) # disable input audio at the start session.input.set_audio_enabled(False) # Set up metrics collection for cost tracking and event logging logger.info("Enabling metrics collection and cost tracking...") session.on("metrics_collected", on_metrics_collected) ctx.add_shutdown_callback(log_session_usage) # Add shutdown callback to log final usage summary @ctx.room.local_participant.register_rpc_method("start_turn") async def start_turn(data: rtc.RpcInvocationData): session.interrupt() session.clear_user_turn() # listen to the caller if multi-user room_io.set_participant(data.caller_identity) session.input.set_audio_enabled(True) @ctx.room.local_participant.register_rpc_method("end_turn") async def end_turn(data: rtc.RpcInvocationData): session.input.set_audio_enabled(False) session.commit_user_turn() @ctx.room.local_participant.register_rpc_method("cancel_turn") async def cancel_turn(data: rtc.RpcInvocationData): session.input.set_audio_enabled(False) session.clear_user_turn() logger.info("cancel turn")

Thanks — you’ve shown the session startup flow, which reveals the issue: you create

session = AgentSession[...]

without passing an

llm

. That makes

session

valid but

session.llm

None. What to do • Pass an LLM (model id or LLM instance) when constructing the session. Example:

session = AgentSession[UserData](
    userdata=userdata,
    turn_detection="manual",
    llm="openai/gpt-4.1-mini",  # or llm=inference.LLM(model="openai/gpt-4.1-mini")
)

• Then start the agent as you already do:

agent = MyAgent(instructions=assembled_instructions, conversation_logger=conversation_logger)
await session.start(agent=agent)

Quick checks • If you already set an LLM elsewhere, confirm it’s set before

await session.start(...)

by printing

session.llm

. • Ensure no code later overwrites

session.llm = None

or your agent class removes it. If you want, paste how you construct MyAgent (its init) or any code that might touch

session.llm

and I’ll point out exactly where to set or preserve the LLM.

'LLM' object has no attribute 'chat_once'

Thanks — you're encountering two things: there is no

chat_once

method on the LLM, and the stream yields chunks with different fields than OpenAI-style

choices

. Here's a concise, working pattern and quick checks. Why the error happens • The LLM only exposes

chat

(streaming).

chat_once

does not exist. • Chunks are

ChatChunk

objects; they often expose

token

,

text

, or

delta.content

— not

choices

. Working async streaming snippet (safe extraction):

temp_chat_ctx = ChatContext()
temp_chat_ctx.items.append(ChatMessage(role="user", content=[prompt]))

response_text = ""
async with context.session.llm.chat(chat_ctx=temp_chat_ctx) as stream:
    async for chunk in stream:
        # try common fields in order
        token = getattr(chunk, "token", None) or getattr(chunk, "text", None)
        if not token:
            # delta may be present
            token = getattr(getattr(chunk, "delta", None), "content", None)
        # final fallback: string representation
        token = token or str(chunk)
        response_text += token or ""

Alternative (non-streaming feel) • If you want a single-response style, still use

chat

but accumulate and stop after first chunk or after stream completes. There's no direct

chat_once

. Quick diagnostics if still broken • Print a single chunk to see real structure:

async with context.session.llm.chat(chat_ctx=temp_chat_ctx) as stream:
    async for chunk in stream:
        print(repr(chunk))
        break

Share that output and I’ll point exactly which field to read. Do you want the snippet adapted to stop after the first complete response (non-streaming behavior) or to stream tokens to downstream components (TTS, UI)? Sources: Pipeline nodes and hooks | LiveKit docs | Large language models (LLM) | LiveKit Docs | Modifying LLM output before TTS | LiveKit docs | LLM-Powered Content Filter | LiveKit docs | Simple Content Filter | LiveKit docs

how to access raw chat context

You're asking how to access the raw chat context. Here’s how to read, copy, and modify it in your tool/agent. Access the current context • From a tool with a RunContext: use

context.session.chat_ctx

. • From inside an Agent instance: use

self.session.chat_ctx

. Inspect messages

chat_ctx = context.session.chat_ctx
for msg in chat_ctx.messages:
    # msg is a ChatMessage object
    print(msg.role, msg.get_text())  # get_text() safely extracts text parts

Create a new or temporary context

from livekit.agents import ChatContext, ChatMessage

# fresh context
temp = ChatContext()
temp.add_message(role="user", content="Hello")

# or copy existing and append
copied = context.session.chat_ctx.copy()
copied.add_message(role="user", content="Follow-up question")

Use a ChatContext with the LLM (streaming

chat

)

async with context.session.llm.chat(chat_ctx=copied) as stream:
    full = ""
    async for chunk in stream:
        token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                or getattr(getattr(chunk, "delta", None), "content", None) \
                or str(chunk)
        full += token or ""

Persist changes to session (if you modify history and want it kept) • After modifying a context in an agent, call your session update method (e.g.,

await self.update_chat_ctx(chat_ctx)

) if your flow requires persisting changes beyond the current turn. Quick notes • Use

.copy()

to avoid mutating the live session context when building temporary prompts. •

ChatMessage

content can be multiparts; use

get_text()

or inspect fields to extract raw text. • If you want me to tailor an example for your tool function, paste the function signature and whether you want a fresh context or history+prompt. Sources: Large language models (LLM) | LiveKit Docs | Tool definition and use | LiveKit docs | Vision | LiveKit docs | Pipeline nodes and hooks | LiveKit docs | Agents and handoffs | LiveKit docs

Traceback (most recent call last): File "/opt/anaconda3/envs/venv/lib/python3.13/site-packages/livekit/agents/voice/generation.py", line 526, in _traceable_fnc_tool val = await function_callable() ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/main_ptt.py", line 122, in conversation_spark return await _conversation_spark(self.session.chat_ctx, context) ^^^^^^^^^^^^^^^^^^^^^ AttributeError: 'AgentSession' object has no attribute 'chat_ctx' {"function": "conversation_spark", "speech_id": "speech_ad2a5549673c", "pid": 41776, "job_id": "AJ_ghRzRF7kaS8T"} is chat_ctx in llm_node stored in the session or agent? why don't have access to it?

You're seeing the AttributeError because

chat_ctx

is not on

AgentSession

. It's owned by the active agent and injected into the RunContext for tools. What to change • In a function tool, use

context.chat_ctx

(the RunContext) — not

self.session.chat_ctx

or

session.chat_ctx

. • If you're inside agent methods and have a bound agent instance, use

self.chat_ctx

. Fix for your traceback Replace the call that does:

return await _conversation_spark(self.session.chat_ctx, context)

with:

return await _conversation_spark(context.chat_ctx, context)

(or

self.chat_ctx

if the tool is executed as a bound agent method and

self

is the agent). Quick checks • Print

context

right before the call to confirm it has `chat_ctx`:

print("run context:", context)
print("has chat_ctx:", hasattr(context, "chat_ctx"), getattr(context, "chat_ctx", None))

• If

context.chat_ctx

is None, share how the agent is started and the tool is invoked so I can point out where the agent/chat context is not being set. If you want, paste the

_conversation_spark

signature and the tool decorator/signature and I’ll show the exact one-line change. Sources: Agents and handoffs | LiveKit docs | LLM-Powered Content Filter | LiveKit docs

for msg in context.chat_ctx.messages[-6] # Last 6 messages for context ^^^^^^^^^^^^^^^^ AttributeError: 'RunContext' object has no attribute 'chat_ctx' {"function": "conversation_spark", "speech_id": "speech_6ebcd6138564", "pid": 57421, "job_id": "AJ_B7mZLP6yMxqi"}

Thanks — you're seeing that error because in your environment

RunContext

doesn't have

chat_ctx

directly. Use

context.session.chat_ctx

inside your tool. Fix example (use this in your function tool):

@function_tool()
async def conversation_spark(context: RunContext):
    # get last 6 messages safely
    chat_ctx = context.session.chat_ctx
    recent = chat_ctx.messages[-6:]
    for msg in recent:
        print(msg.role, msg.get_text())

    # build a temporary context if you want to avoid mutating live history
    temp = chat_ctx.copy()
    temp.add_message(role="user", content=prompt)

    # stream from LLM and assemble text
    response_text = ""
    async with context.session.llm.chat(chat_ctx=temp) as stream:
        async for chunk in stream:
            token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                    or getattr(getattr(chunk, "delta", None), "content", None) \
                    or str(chunk)
            response_text += token or ""

    return response_text.strip()

Quick checks: • Confirm your tool signature is

async def fn(context: RunContext, ...)

. • Print

print("has session:", hasattr(context, "session"), "has chat_ctx:", hasattr(getattr(context, "session", None), "chat_ctx"))

if you still hit errors. • If

context.session.chat_ctx

is None, share how/where you start the session and agent so I can point out why chat context isn't initialized. Which behavior do you want: use history+prompt or a fresh prompt-only context? Sources: Tool definition and use | LiveKit docs | External data and RAG | LiveKit docs | Agents and handoffs | LiveKit docs

but for msg in context.session.chat_ctx.messages[-6] # Last 6 messages for context ^^^^^^^^^^^^^^^^^^^^^^^^ AttributeError: 'AgentSession' object has no attribute 'chat_ctx' {"function": "conversation_spark", "speech_id": "speech_33f366053bea", "pid": 64443, "job_id": "AJ_NfMb2vhMF7rL"}

The error means you're looking in the wrong place. The active agent holds the chat context — not AgentSession directly. Use this inside a function tool:

# correct: access agent's chat_ctx via session.agent
chat_ctx = context.session.agent.chat_ctx
for msg in chat_ctx.messages[-6:]:
    print(msg.role, msg.get_text())

Or, if the tool is a bound agent method, use `self.chat_ctx`:

# inside an Agent method
for msg in self.chat_ctx.messages[-6:]:
    ...

Quick debugging prints to confirm what you have at runtime:

print("has session:", hasattr(context, "session"))
print("has agent on session:", hasattr(context.session, "agent"))
print("has chat_ctx on agent:", hasattr(getattr(context.session, "agent", None), "chat_ctx"))

If

context.session.agent

is missing or

agent.chat_ctx

is None, share how you start the session and agent (the

await session.start(agent=...)

call) and the agent class constructor — I’ll point out where the agent/chat context is not being set. Sources: Agents and handoffs | LiveKit docs | Pipeline nodes and hooks | LiveKit docs | External data and RAG | LiveKit docs

async def entrypoint(ctx: JobContext): # Initialize conversation logger conversation_logger = ConversationLogger(enabled=True) if TEST_MODE else None # Create user data and session (before connection to avoid audio buffer warning) userdata = UserData(session_config=None) session = AgentSession[UserData]( userdata=userdata, llm="openai/gpt-4.1-mini", turn_detection="manual", ) # Create and start RoomIO before connecting room_io = RoomIO(session, room=ctx.room) await room_io.start() # Now connect to the room await ctx.connect() logger.info(f"Session started - Room: {ctx.room.name}") # Wait for participant to join logger.info("Waiting for participant to join...") participant = await ctx.wait_for_participant() logger.info(f"Participant joined: {participant.identity}") # Get session configuration if USE_DEFAULT_TEST_CONFIG: logger.info("Using default session config for test environment") session_config = get_default_session_config() else: session_config = await get_session_config(ctx.room, participant.identity) print("------------------ session_config -------------------") print(session_config) conversation_logger.log_message("system", session_config) if conversation_logger else None print("------------------------------------------------------") if not session_config: logger.warning("Failed to retrieve session config, using defaults") session_config = get_default_session_config() # Update user data with session config userdata.session_config = session_config # Assemble prompt from session config assembled_instructions = assemble_prompt_from_config(session_config) logger.info("Assembled instructions from session config") if TEST_MODE: conversation_logger.log_message("system", assembled_instructions) # Create agent and start session agent = MyAgent(instructions=assembled_instructions, conversation_logger=conversation_logger) await session.start(agent=agent) # disable input audio at the start session.input.set_audio_enabled(False) # Set up metrics collection for cost tracking and event logging logger.info("Enabling metrics collection and cost tracking...") session.on("metrics_collected", on_metrics_collected) ctx.add_shutdown_callback(log_session_usage) # Add shutdown callback to log final usage summary @ctx.room.local_participant.register_rpc_method("start_turn") async def start_turn(data: rtc.RpcInvocationData): session.interrupt() session.clear_user_turn() # listen to the caller if multi-user room_io.set_participant(data.caller_identity) session.input.set_audio_enabled(True) @ctx.room.local_participant.register_rpc_method("end_turn") async def end_turn(data: rtc.RpcInvocationData): session.input.set_audio_enabled(False) session.commit_user_turn() @ctx.room.local_participant.register_rpc_method("cancel_turn") async def cancel_turn(data: rtc.RpcInvocationData): session.input.set_audio_enabled(False) session.clear_user_turn() logger.info("cancel turn")

Thanks — I can see the session startup you posted. Two quick facts and a minimal fix so your tool stops raising AttributeError: 1) Where chat context lives • The active Agent owns the chat context. In tools the path is:

context.session.agent.chat_ctx

. •

AgentSession

itself does not have

chat_ctx

. 2) Minimal fix for your tool Replace any use of

context.session.chat_ctx

or

self.session.chat_ctx

with

context.session.agent.chat_ctx

. Example:

@function_tool()
async def conversation_spark(context: RunContext):
    chat_ctx = context.session.agent.chat_ctx
    recent = chat_ctx.messages[-6:]
    # build temporary copy to avoid mutating live history
    temp = chat_ctx.copy()
    temp.add_message(role="user", content=[prompt])

    response_text = ""
    async with context.session.llm.chat(chat_ctx=temp) as stream:
        async for chunk in stream:
            token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                    or getattr(getattr(chunk, "delta", None), "content", None) \
                    or str(chunk)
            response_text += token or ""

    return response_text.strip()

Quick debug prints to confirm runtime shape:

print("has session:", hasattr(context, "session"))
print("has agent on session:", hasattr(context.session, "agent"))
print("has chat_ctx on agent:", hasattr(getattr(context.session, "agent", None), "chat_ctx"))

If

context.session.agent

is missing or

agent.chat_ctx

is None, paste how you construct MyAgent (its init) and the

await session.start(agent=agent)

call — I’ll point out exactly where the agent/chat context isn’t being set.

2025-12-22 172042,687 - ERROR livekit.agents - exception occurred while executing tool Traceback (most recent call last): File "/opt/anaconda3/envs/venv/lib/python3.13/site-packages/livekit/agents/voice/generation.py", line 526, in _traceable_fnc_tool val = await function_callable() ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/main_ptt.py", line 122, in conversation_spark return await _conversation_spark(context) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/session_tools.py", line 36, in conversation_spark for msg in context.session.agent.chat_ctx.messages[-6] # Last 6 messages for context ^^^^^^^^^^^^^^^^^^^^^ AttributeError: 'AgentSession' object has no attribute 'agent' {"function": "conversation_spark", "speech_id": "speech_1a8fe737bd49", "pid": 70759, "job_id": "AJ_cUZ9fy3vrDQe"}

Thanks — the traceback shows

context.session

at runtime isn't populated with an

agent

attribute, so

context.session.agent.chat_ctx

fails. Quick targeted checks and fixes: 1) Confirm session.start completed before tools run • Ensure

await session.start(agent=agent)

finished before any RPCs or tools invoke

conversation_spark

. Tools called before the agent is attached will see an incomplete session. 2) Print runtime shape immediately in the tool to see what's available:

print("context:", type(context))
print("has session:", hasattr(context, "session"))
print("session type:", type(getattr(context, "session", None)))
print("has agent on session:", hasattr(getattr(context, "session", None), "agent"))
print("session attrs:", vars(getattr(context, "session", None)) if getattr(context, "session", None) else None)

Share that output if you want me to interpret it. 3) Safe access pattern (avoid AttributeError)

sess = getattr(context, "session", None)
agent = getattr(sess, "agent", None)
if agent is None:
    # bail or handle gracefully
    logger.warning("Agent not attached to session yet")
    return ""
chat_ctx = agent.chat_ctx
recent = chat_ctx.messages[-6:]

4) Common root causes • Tools invoked before

session.start(...)

completes (race condition). • You started a plain AgentSession but never attached/started your agent instance properly. • Custom agent/session subclass overrides that remove

agent

reference. If you paste the output from the debug prints above and confirm when

conversation_spark

is called relative to

await session.start(...)

, I’ll point to the exact line to change or the ordering bug to fix. Sources: Agents (Python) | livekit/livekit_composite | DeepWiki | Agents and handoffs | LiveKit docs

=== DEBUG conversation_spark === context: <class 'livekit.agents.voice.events.RunContext'> has session: True session type: <class 'livekit.agents.voice.agent_session.AgentSession'> session attrs: ['__aenter__', '__aexit__', '__annotations__', '__class__', '__class_getitem__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__firstlineno__', '__format__', '__ge__', '__getattribute__', '__getstate__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__orig_bases__', '__orig_class__', '__parameters__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__static_attributes__', '__str__', '__subclasshook__', '__weakref__', '_aclose_impl', '_activity', '_activity_lock', '_agent', '_agent_speaking_span', '_agent_state', '_cancel_user_away_timer', '_chat_ctx', '_close_soon', '_closing', '_closing_task', '_conn_options', '_conversation_item_added', '_events', '_forward_audio_atask', '_forward_audio_task', '_forward_video_atask', '_forward_video_task', '_global_run_state', '_input', '_job_context_cb_registered', '_llm', '_llm_error_counts', '_lock', '_loop', '_mcp_servers', '_next_activity', '_on_audio_input_changed', '_on_audio_output_changed', '_on_error', '_on_text_output_changed', '_on_video_input_changed', '_on_video_output_changed', '_opts', '_output', '_room_io', '_root_span_context', '_session_span', '_set_user_away_timer', '_started', '_stt', '_tool_items_added', '_tts', '_tts_error_counts', '_turn_detection', '_update_activity', '_update_activity_atask', '_update_activity_task', '_update_agent_state', '_update_user_state', '_user_away_timer', '_user_input_transcribed', '_user_speaking_span', '_user_state', '_userdata', '_vad', '_video_sampler', 'aclose', 'agent_state', 'clear_user_turn', 'commit_user_turn', 'conn_options', 'current_agent', 'current_speech', 'drain', 'emit', 'generate_reply', 'history', 'input', 'interrupt', 'llm', 'mcp_servers', 'off', 'on', 'once', 'options', 'output', 'run', 'say', 'shutdown', 'start', 'stt', 'tts', 'turn_detection', 'update_agent', 'update_options', 'user_state', 'userdata', 'vad'] has chat_ctx: False has agent: False

Thanks — that debug output is very helpful. What’s happening • Your AgentSession instance has internal

_chat_ctx

and

_agent

-related fields, but no public

agent

attribute. However it exposes

current_agent

and

llm

publicly. • So the safe, supported way to get the active chat context in your tool is via

context.session.current_agent.chat_ctx

(not

context.session.agent

or

context.session.chat_ctx

). Minimal fix Replace your access with:

sess = getattr(context, "session", None)
agent = getattr(sess, "current_agent", None)
if not agent:
    logger.warning("No active agent on session yet")
    return ""
chat_ctx = agent.chat_ctx
for msg in chat_ctx.messages[-6:]:
    ...

Safer streaming example using a temp copy:

chat_ctx = agent.chat_ctx
temp = chat_ctx.copy()
temp.add_message(role="user", content=[prompt])

response_text = ""
async with sess.llm.chat(chat_ctx=temp) as stream:
    async for chunk in stream:
        token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                or getattr(getattr(chunk, "delta", None), "content", None) \
                or str(chunk)
        response_text += token or ""
return response_text.strip()

Debug tips • If

current_agent

is None, tools are being invoked before

await session.start(agent=...)

completes or the agent was never attached. • Print

print("current_agent:", getattr(sess, "current_agent", None))

right before use to confirm. • Avoid accessing private

_chat_ctx

unless you must — use

current_agent.chat_ctx

. If you want, paste the exact tool decorator/signature and where the RPC is queued relative to

await session.start(...)

and I’ll point out the one-line ordering change to guarantee

current_agent

is set before the tool runs.

how would you rewrite this? @function_tool() async def conversation_spark( context: RunContext, ) -> str: """ Generate an unexpected question or observation to shift conversation energy. Call this when: - Conversation feels flat or stuck in surface-level responses - User gives 3+ short responses in a row - You want to build rapport through genuine curiosity - After 60-90 seconds to add variety and depth Returns: A specific, contextual question or scenario that's interesting and reveals thinking patterns """ logger.info("Generating conversation spark...") # Access session and get current agent's chat context sess = getattr(context, "session", None) if sess is None: logger.warning("Session not available in context") return "If you could change one thing about how today went, what would it be?" # Get agent and its chat_ctx agent = getattr(sess, "current_agent", None) if agent is None: logger.warning("Agent not available in session") return "If you could change one thing about how today went, what would it be?" chat_ctx = getattr(agent, "chat_ctx", None) if chat_ctx is None: logger.warning("chat_ctx not available - cannot access conversation history") return "If you could change one thing about how today went, what would it be?" print("chat_ctx: ", chat_ctx) prompt = f"""You are helping a voice AI have a natural, engaging conversation. Recent conversation: {conversation_history} Generate ONE specific question or scenario that will: 1. Be genuinely interesting to think about (not generic) 2. Reveal how the person thinks/feels naturally 3. NOT sound like therapy ("tell me more", "how does that make you feel") 4. Shift energy if conversation feels flat 5. Be conversational and friend-like Examples of GOOD questions: - "If you could go back and redo one conversation from this week, which one?" - "What kind of stress - like drowning in tasks, or more the 'am I even good at this' kind?" - "If you could automate one annoying part of your job, what would it be?" Examples of BAD questions (avoid these): - "Tell me more about that" - "How does that make you feel?" - "Can you elaborate?" Return ONLY the question/observation, nothing else. Keep it under 25 words.""" try: # Use the session's LLM to generate a contextual spark response_text = "" async with context.session.llm.chat(chat_ctx=chat_ctx) as stream: async for chunk in stream: # Try common fields in order token = getattr(chunk, "token", None) or getattr(chunk, "text", None) if not token: # delta may be present delta = getattr(chunk, "delta", None) token = getattr(delta, "content", None) if delta else None # Final fallback: string representation token = token or str(chunk) response_text += token or "" spark = response_text.strip() logger.info(f"Generated conversation spark: {spark}") return spark except Exception as e: logger.error(f"Error generating conversation spark: {e}") # Fallback to a generic but good question return "If you could change one thing about how today went, what would it be?"

Below is a cleaned, robust rewrite of your tool. It safely accesses the active agent's chat context, builds a temporary prompt (copies history), streams the LLM response with resilient token extraction, and returns a short question fallback on errors.

from livekit.agents import function_tool, RunContext, ChatContext, ChatMessage

@function_tool()
async def conversation_spark(context: RunContext) -> str:
    """
    Generate one short, curiosity-provoking question to shift conversation energy.
    Returns a single question (<= 25 words). Falls back to a safe default on error.
    """
    <http://logger.info|logger.info>("Generating conversation spark...")

    # Safe session / agent access
    sess = getattr(context, "session", None)
    if not sess:
        logger.warning("Session not available in context")
        return "If you could change one thing about how today went, what would it be?"

    agent = getattr(sess, "current_agent", None)
    if not agent:
        logger.warning("No active agent on session")
        return "If you could change one thing about how today went, what would it be?"

    chat_ctx = getattr(agent, "chat_ctx", None)
    if chat_ctx is None:
        logger.warning("chat_ctx not available on agent")
        return "If you could change one thing about how today went, what would it be?"

    # Build a temporary context (copy last N messages to avoid mutating live history)
    recent = chat_ctx.messages[-6:]
    temp_ctx = chat_ctx.copy()
    # trim to last 6 messages if copy preserves full history
    temp_ctx.messages = list(recent)

    # Compose instruction + recent conversation for the LLM
    conversation_history = "\n".join(f"{m.role}: {m.get_text()}" for m in recent)
    system_instructions = (
        "You are helping a voice AI produce ONE specific, conversational question (<=25 words) "
        "that: is interesting, reveals thinking, avoids therapy prompts, and can shift flat energy. "
        "Return ONLY the question."
    )
    temp_ctx.add_message(role="system", content=system_instructions)
    temp_ctx.add_message(role="user", content=f"Recent conversation:\n{conversation_history}")

    # Stream from LLM and assemble text robustly
    try:
        response_text = ""
        async with sess.llm.chat(chat_ctx=temp_ctx) as stream:
            async for chunk in stream:
                # resilient extraction of token/text/delta
                token = getattr(chunk, "token", None) or getattr(chunk, "text", None)
                if not token:
                    delta = getattr(chunk, "delta", None)
                    token = getattr(delta, "content", None) if delta else None
                token = token or (str(chunk) if chunk is not None else "")
                response_text += token or ""

        spark = response_text.strip()
        logger.info(f"Generated conversation spark: {spark}")
        # final safety: return fallback if empty or too long
        if not spark:
            raise ValueError("Empty LLM response")
        if len(spark.split()) > 30:  # generous safeguard vs 25 words requirement
            spark = "If you could change one thing about how today went, what would it be?"
        return spark

    except Exception as e:
        logger.error(f"Error generating conversation spark: {e}")
        return "If you could change one thing about how today went, what would it be?"

If you want, I can: • Trim the system/user prompt to make the LLM more likely to obey the 25-word limit. • Return streaming tokens to TTS/UI instead of accumulating first. Which do you prefer?

look in the docs plz Traceback (most recent call last): File "/opt/anaconda3/envs/venv/lib/python3.13/site-packages/livekit/agents/voice/generation.py", line 526, in _traceable_fnc_tool val = await function_callable() ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/main_ptt.py", line 122, in conversation_spark return await _conversation_spark(context) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/session_tools.py", line 35, in conversation_spark msgs = chat_ctx.get_messages() ^^^^^^^^^^^^^^^^^^^^^ AttributeError: '_ReadOnlyChatContext' object has no attribute 'get_messages'. Did you mean: 'add_message'? {"function": "conversation_spark", "speech_id": "speech_6a499e693438", "pid": 40498, "job_id": "AJ_iM46ptJ2bLjM"}

Thanks — I checked the docs and your runtime errors. Summary and exact fix: • The read-only chat context passed into tools is a _ReadOnlyChatContext. It does not have get_messages() or allow mutation methods that would change live history. • To read history in a tool, use the

.messages

property on that context object (each item is a ChatMessage). Do not call get_messages(). Minimal corrected pattern for your tool (focus on the failing lines):

# inside your function tool
# get the read-only chat context from the active agent
sess = getattr(context, "session", None)
agent = getattr(sess, "current_agent", None) or getattr(sess, "agent", None)
if not agent or not getattr(agent, "chat_ctx", None):
    logger.warning("No active agent/chat_ctx")
    return "If you could change one thing about how today went, what would it be?"

chat_ctx = agent.chat_ctx             # read-only context object
msgs = chat_ctx.messages              # <-- correct: use .messages, not get_messages()

recent = msgs[-6:] if msgs else []
conversation_history = "\n".join(f"{m.role}: {m.get_text()}" for m in recent)

When calling the LLM, build a temporary ChatContext (so you don't mutate the read-only one), add system/user messages, then stream:

temp = ChatContext()
temp.add_message(role="system", content=system_instructions)
temp.add_message(role="user", content=conversation_history)

response_text = ""
async with sess.llm.chat(chat_ctx=temp) as stream:
    async for chunk in stream:
        token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                or getattr(getattr(chunk, "delta", None), "content", None) \
                or str(chunk)
        response_text += token or ""

Quick checklist to avoid similar errors: • Use

context.session.current_agent.chat_ctx

(or

context.session.agent.chat_ctx

if present) to get the chat context. • Read history via

.messages

. • Copy into a new

ChatContext()

for LLM calls; do not mutate the read-only object. • Use resilient extraction for streamed chunks (token/text/delta). If you want, paste the exact few lines around where you build

conversation_history

and I’ll patch them inline into your function. Sources: External data and RAG | LiveKit docs | Tool definition and use | LiveKit docs

File "/Users/emilyli/LocalDev/inner-call-backend/innercall-ptt/session_tools.py", line 35, in conversation_spark msgs = chat_ctx.messages ^^^^^^^^^^^^^^^^^ AttributeError: '_ReadOnlyChatContext' object has no attribute 'messages' {"function": "conversation_spark", "speech_id": "speech_ceb83cbde24c", "pid": 76614, "job_id": "AJ_pMN3SLmQa9Cu"} nor does messages() work

Thanks — I dug through the docs and your runtime errors. The core issue: the chat context object you receive in tools is a read-only wrapper and its API varies by LiveKit version, so direct accessors you tried (

.messages

,

get_messages()

,

get_messages

) may not exist. What to do now — pick one of these reliable approaches and apply it exactly: 1) Use session.history (recommended, works across versions) • Access full conversation items from the session:

history = context.session.history  # list of conversation items
recent_msgs = [it for it in history if getattr(it, "type", None) == "message"][-6:]
conversation_history = "\n".join(f"{m.role}: {m.text_content}" for m in recent_msgs)

2) If you need the chat context object specifically, iterate the read-only context (if supported) or copy it • Try iterating or copying (some versions support this):

# attempt 1: iterate (async or sync depending on version)
msgs = [m for m in context.chat_ctx]  # or: [m async for m in context.chat_ctx]

# attempt 2: copy to mutable ChatContext and inspect
copied = context.chat_ctx.copy()
msgs = copied.messages  # now accessible on the copied object

3) Persist a mutable history yourself (most robust) • Maintain a simple list in

session.userdata

when messages arrive, then read it in tools:

# when handling new messages elsewhere:
context.session.userdata.messages.append({"role": role, "text": text})

# in tool:
msgs = context.session.userdata.messages[-6:]

Minimal working replacement for your tool using session.history (copy this into your function):

sess = getattr(context, "session", None)
if not sess:
    return "If you could change one thing about how today went, what would it be?"

history = getattr(sess, "history", []) or []
recent = [it for it in history if getattr(it, "type", None) == "message"][-6:]
conversation_history = "\n".join(f"{m.role}: {getattr(m, 'text_content', getattr(m, 'text', ''))}" for m in recent)

# build temp ChatContext for LLM
temp = ChatContext()
temp.add_message(role="system", content=system_instructions)
temp.add_message(role="user", content=conversation_history)

# stream from LLM
response_text = ""
async with sess.llm.chat(chat_ctx=temp) as stream:
    async for chunk in stream:
        token = getattr(chunk, "token", None) or getattr(chunk, "text", None) \
                or getattr(getattr(chunk, "delta", None), "content", None) \
                or str(chunk)
        response_text += token or ""

return (response_text.strip() or "If you could change one thing about how today went, what would it be?")

If you want, tell me: • Your LiveKit Agents package version (output of

pip show livekit-agents

), and • Whether

context.chat_ctx.copy()

works (you can try quickly and paste the exception/output). With that I’ll give the exact one-line fix tailored to your version. Sources: External data and RAG | LiveKit docs | Agents and handoffs | LiveKit docs

Name: livekit-agents Version: 1.2.18 Summary: A powerful framework for building realtime voice AI agents Home-page: Author: Author-email: LiveKit <hello@livekit.io> License: Location: /opt/anaconda3/envs/venv/lib/python3.13/site-packages Requires: aiohttp, av, certifi, click, colorama, docstring-parser, eval-type-backport, livekit, livekit-api, livekit-blingfire, livekit-protocol, nest-asyncio, numpy, openai, opentelemetry-api, opentelemetry-exporter-otlp, opentelemetry-sdk, prometheus-client, protobuf, psutil, pydantic, pyjwt, sounddevice, types-protobuf, typing-extensions, watchfiles Required-by: livekit-plugins-anthropic, livekit-plugins-aws, livekit-plugins-cartesia, livekit-plugins-deepgram, livekit-plugins-elevenlabs, livekit-plugins-google, livekit-plugins-openai, livekit-plugins-silero, livekit-plugins-turn-detector

Thanks — with LiveKit Agents 1.2.18 the correct, reliable source of conversation history in tools is

context.session.history

. The read-only chat context passed into tools doesn't expose

.messages

or

.get_messages()

, so use

session.history

, pick the last message items you need, copy them into a temporary

ChatContext

, then call

session.llm.chat(...)

. Minimal, version-compatible snippet you can drop into your tool:

from livekit.agents import function_tool, RunContext, ChatContext

@function_tool()
async def conversation_spark(context: RunContext) -> str:
    <http://logger.info|logger.info>("Generating conversation spark...")

    sess = getattr(context, "session", None)
    if not sess:
        return "If you could change one thing about how today went, what would it be?"

    # session.history is the supported source of message items
    history = getattr(sess, "history", []) or []
    recent = [it for it in history if getattr(it, "type", None) == "message"][-6:]

    # Build a temporary ChatContext from recent history
    temp = ChatContext()
    for item in recent:
        text = getattr(item, "text_content", None) or getattr(item, "text", "") or str(getattr(item, "content", ""))
        temp.add_message(role=getattr(item, "role", "user"), content=text)

    # Add instructions for the spark
    system_instructions = (
        "You are helping a voice AI produce ONE specific, conversational question (<=25 words) "
        "that is interesting, reveals thinking, avoids therapy-style prompts, and can shift flat energy. "
        "Return ONLY the question."
    )
    temp.add_message(role="system", content=system_instructions)

    # Stream from the session's LLM and assemble text robustly
    try:
        response_text = ""
        async with sess.llm.chat(chat_ctx=temp) as stream:
            async for chunk in stream:
                token = getattr(chunk, "token", None) or getattr(chunk, "text", None)
                if not token:
                    delta = getattr(chunk, "delta", None)
                    token = getattr(delta, "content", None) if delta else None
                response_text += token or ""
        spark = response_text.strip()
        if not spark:
            raise ValueError("Empty LLM response")
        if len(spark.split()) > 30:
            return "If you could change one thing about how today went, what would it be?"
        return spark
    except Exception as e:
        logger.error("Error generating conversation spark: %s", e)
        return "If you could change one thing about how today went, what would it be?"

If you want, I can: • Shorten the system prompt to increase adherence to the 25-word limit. • Return streaming tokens to TTS instead of waiting for the full response. Which would you prefer? Sources: External data and RAG | LiveKit docs | Large language models (LLM) | LiveKit Docs