Hey, I'm trying to send metadata from the nextjs app to the python worker and haven't been able to do it yet. This is what I'm sending in my nextjs app:

[createParticipantToken] RoomConfiguration creado: {
  "name": "",
  "emptyTimeout": 0,
  "departureTimeout": 0,
  "maxParticipants": 0,
  "minPlayoutDelay": 0,
  "maxPlayoutDelay": 0,
  "syncStreams": false,
  "agents": [],
  "metadata": "{\"user_id\":\"user_123\",\"user_name\":\"Francisco\",\"integrations\":{\"google_maps\":{\"enabled\":true,\"token\":\"ey...\"}}}"                                                                                                                    :\"ey...\"}}}"        
}

And this is my worker:

import logging
import json

from dotenv import load_dotenv
from livekit import rtc
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    MetricsCollectedEvent,
    FunctionToolsExecutedEvent,
    cli,
    room_io,
)
from livekit.plugins import noise_cancellation
from livekit.plugins.turn_detector.multilingual import MultilingualModel
from observability.langfuse import flush_traces, setup_langfuse
from observability.metrics import LatencyTracker
from observability.usage import UsageTracker
from audio.stt import build_stt
from audio.tts import build_tts
from audio.vad import load_vad
from llm.llm import build_llm
from prompts.loader import load_prompt
from mcp_clients.loader import load_mcp_servers_from_metadata
from tools.loader import register_tools


logger = logging.getLogger("agent")

load_dotenv(".env.local")

@register_tools
class Assistant(Agent):
    def __init__(self, user_name: str) -> None:
        base_prompt = load_prompt("wil")
        formatted_instructions = base_prompt.replace("{{user_name}}", user_name)

        super().__init__(
            instructions=formatted_instructions,
        )


server = AgentServer()


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = load_vad()


server.setup_fnc = prewarm


@server.rtc_session()
async def my_agent(ctx: JobContext):
    # Logging setup
    # Add any other context you want in all log entries here
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }

    # Setup Langfuse tracing
    trace_provider = setup_langfuse(
        metadata={
            "langfuse.session.id": ctx.room.name,
            "agent": "assistant",
            "lang": "es",
        }
    )

    # Callback to flush traces when the session ends
    if trace_provider:
        async def flush_langfuse_traces():
            flush_traces(trace_provider)

        ctx.add_shutdown_callback(flush_langfuse_traces)
    
    user_name = "Usuario" 
    active_mcp_servers = []

    try:
        if ctx.job.metadata:
            meta_data = json.loads(ctx.job.metadata)

            user_name = meta_data.get("user_name", "Usuario")
            <http://logger.info|logger.info>(f"User context: {user_name}")

            active_mcp_servers = load_mcp_servers_from_metadata(ctx.job.metadata)
            
    except Exception as e:
        logger.error(f"Error parsing metadata: {e}")

    # Set up a voice AI pipeline using OpenAI, Cartesia, AssemblyAI, and the LiveKit turn detector
    session = AgentSession(
        stt=build_stt(language="es"),
        llm=build_llm(),
        tts=build_tts(language="es"),

        turn_detection=MultilingualModel(),
        mcp_servers=active_mcp_servers,
        
        vad=ctx.proc.userdata["vad"],
        max_endpointing_delay=1.0,
        min_endpointing_delay=0.3,
        min_interruption_duration=0.2,
        # allow the LLM to generate a response while waiting for the end of turn
        # See more at <https://docs.livekit.io/agents/build/audio/#preemptive-generation>
        preemptive_generation=True,
    )

    latency = LatencyTracker()
    usage = UsageTracker()

    @session.on("metrics_collected")
    def on_metrics(ev: MetricsCollectedEvent):
        m = ev.metrics
        logger.debug(
            f"metrics_collected: type={type(m).__name__}"
        )
        latency.collect(m)
        usage.collect(m)

    @session.on("function_tools_executed")
    def on_tools_executed(event: FunctionToolsExecutedEvent):
        <http://logger.info|logger.info>("🔧 Tools executed")

        for call, output in event.zipped():
            <http://logger.info|logger.info>(f"Tool name: {call.name}")

            # El output suele ser dict / list / str
            try:
                <http://logger.info|logger.info>(
                    "Tool output:\n%s",
                    json.dumps(output.output, indent=2, ensure_ascii=False)
                )
            except Exception:
                <http://logger.info|logger.info>(f"Tool output (raw): {output.output}")

    
    async def log_session_metrics():
        latency_summary = latency.summary()

        <http://logger.info|logger.info>(
            f"[room={ctx.room.name}] latency_summary={latency_summary}"
        )

        <http://logger.info|logger.info>(
            f"[room={ctx.room.name}] usage={usage.summary()}"
        )

    ctx.add_shutdown_callback(log_session_metrics)

    ctx.log_context_fields |= {
        "agent": "assistant",
        "lang": "es",
    }
    # Start the session, which initializes the voice pipeline and warms up the models
    await session.start(
        agent=Assistant(user_name=user_name),
        room=ctx.room,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
                if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                else noise_cancellation.BVC(),
            ),
        ),
    )

    # Join the room and connect to the user
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(
        server
    )

What am I missing?

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

what is the difference between room & job metadata? a room has only one job as far as I understand. and a job belongs to a room. Is that correct?

Good question — short answer first, then the key differences. Short answer • No: a room can have many jobs (agents). A job always belongs to one room. • Use room metadata for shared, room-wide state. Use job (job/dispatch) metadata for agent-specific configuration or per-job data. Differences (concise) • Scope ◦ Room metadata: belongs to the room, visible to all participants and agents in that room (

ctx.room.metadata

). ◦ Job metadata: belongs to a single agent job/dispatch (

ctx.job.metadata

). • Lifecycle ◦ Room metadata: exists for the room’s lifetime (until updated or room deleted). ◦ Job metadata: created when the agent job is dispatched and removed when the job ends. • Use cases ◦ Room metadata: meeting topic, language, shared flags, anything every participant/agent should see. ◦ Job metadata: agent-specific inputs (user id for that agent, API keys, task instructions), different for each agent in the same room. • How to set/access ◦ Room metadata: set on room creation/update or via token; read as

ctx.room.metadata

(call

await ctx.connect()

first to ensure it’s available). ◦ Job metadata: provided when dispatching the agent (API/CLI/token dispatch or RoomConfiguration.agents with explicit dispatch); read as

ctx.job.metadata

inside the agent. Example scenario • Two agents in the same room: a transcription agent and a survey agent. ◦ Shared language → room metadata. ◦ Survey question or target user → job metadata for the survey agent. If you want, tell me how you currently create/dispatch agents (automatic dispatch, API, token, or CLI) and I’ll recommend the exact place to put your metadata and sample code to read it. Sources: Agent dispatch | LiveKit docs | Job lifecycle | LiveKit docs | Survey Calling Agent | LiveKit docs

At the moment i only have one agent. But I wanna send info about the user: id, name, and things the agent or agents would use to address the. if there were many agents they should allsee this data.

Great — since you want user info (id, name, etc.) to be visible to all agents, use room metadata. That makes the data shared across the room and available to every agent and participant. What to do (concise) 1) Send the user object as room metadata from your Next.js backend when creating the room or generating the join token: • Put a JSON string in RoomConfiguration.metadata (or set room metadata via RoomService). 2) In your Python agent, read it from

ctx.room.metadata

after connecting:

await ctx.connect()
if ctx.room.metadata:
    data = json.loads(ctx.room.metadata)
    user_id = data.get("user_id")
    user_name = data.get("user_name")

Notes and tips • Room metadata is shared by all agents and participants. Job metadata is per-agent (use only when you need agent-specific config). • Ensure you call

await ctx.connect()

before reading

ctx.room.metadata

so the room state is available. • Keep the metadata reasonably small (tokens or secrets: prefer passing only references and fetching secrets server-side). If you want, tell me how you create the room/token (code snippet) and I’ll give the exact Next.js/backend code to attach the JSON metadata.

How is it possible that I'm not receiving the room metadata if as I showed you, I'm sending it.

13:50:28.749 DEBUG  livekit.agents     received assignment  
                                         {"agent_name": "", "room_id": "RM_mroc44bKd6Xk", "room": "voice_assistant_room_1439", "job_id": "AJ_SpTn32fVqK6C", 
"dispatch_id": "AD_K5ENoByWsj3y", "enable_recording": false}
    13:50:28.755 INFO   livekit.agents     initializing job runner {"tid": 6324}
    13:50:28.997 INFO   livekit.agents     job runner initialized {"tid": 6324, "elapsed_time": 0.24}
    13:50:29.004 DEBUG  asyncio            Using proactor: IocpProactor  
    13:50:29.013 INFO   observab….langfuse Langfuse tracing configured successfully {"room": "voice_assistant_room_1439"}
    13:50:29.015 INFO   agent              Room metadata available: True, value: '' {"room": "voice_assistant_room_1439"}
    13:50:29.018 INFO   agent              Job metadata available: True, value: '' {"room": "voice_assistant_room_1439"}
    13:50:29.019 INFO   agent              Room name: voice_assistant_room_1439 {"room": "voice_assistant_room_1439"}
    13:50:29.021 WARNI… agent              No metadata found in room or job {"room": "voice_assistant_room_1439"}
    13:50:29.023 INFO   stt                Initializing STT with fallback: Deepgram nova-2 -> Speechmatics {"room": "voice_assistant_room_1439"}
    13:50:29.025 INFO   llm                Initializing LLM with fallback: gpt-4.1-mini -> gpt-oss-120b {"room": "voice_assistant_room_1439"}
    13:50:29.890 INFO   tts                Initializing TTS with fallback: Cartesia -> Inworld {"room": "voice_assistant_room_1439"}
    13:50:29.893 INFO   livekit.agents     resampling livekit.plugins.cartesia.tts.TTS from 24000Hz to 48000Hz {"room": "voice_assistant_room_1439"}
    13:50:30.071 DEBUG  livekit.agents     input stream attached  
                                         {"participant": null, "source": "SOURCE_UNKNOWN", "accepted_sources": ["SOURCE_MICROPHONE"], "room": 
"voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:30.077 DEBUG  livekit.agents     http_session(): creating a new httpclient ctx
                                         {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:30.843 DEBUG  livekit.….cartesia Established new Cartesia TTS WebSocket connection  
                                           {"cartesia_request_id": "2d55a9ff-bfa2-45eb-8a0a-0568fd55c79b", "room": "voice_assistant_room_1439", "agent": 
"assistant", "lang": "es"}
    13:50:31.169 DEBUG  livekit.agents     using audio io: `RoomIO` -> `AgentSession` -> `TranscriptSynchronizer` -> `RoomIO`  
                                         {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:31.171 DEBUG  livekit.agents     using transcript io: `AgentSession` -> `TranscriptSynchronizer` -> `RoomIO`
                                         {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:31.174 INFO   agent              After connect - Room metadata:
                                           '{"user_id":"user_123","user_name":"Francisco","integrations":{"google_maps":{"enabled":true,"token":"ey..."}}}'     
                                {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:31.178 INFO   agent              After connect - Job metadata: '' {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:31.179 INFO   agent              Job attributes: ['ByteSize', 'Clear', 'ClearExtension', 'ClearField', 'CopyFrom', 'DESCRIPTOR',
                                           'DiscardUnknownFields', 'FindInitializationErrors', 'FromString', 'HasExtension', 'HasField', 'IsInitialized',       
                                           'ListFields', 'MergeFrom', 'MergeFromString', 'ParseFromString', 'SerializePartialToString', 'SerializeToString',    
                                           'SetInParent', 'UnknownFields', 'WhichOneof', 'agent_name', 'dispatch_id', 'enable_recording', 'id', 'metadata',     
                                           'namespace', 'participant', 'room', 'state', 'type']
                                {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}
    13:50:31.185 INFO   agent              Job metadata type: <class 'str'>, value: '' {"room": "voice_assistant_room_1439", "agent": "assistant", "lang": "es"}    13:50:31.455 DEBUG  livekit.agents     start reading stream

import logging
import json

from dotenv import load_dotenv
from livekit import rtc
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    MetricsCollectedEvent,
    FunctionToolsExecutedEvent,
    cli,
    room_io,
)
from livekit.plugins import noise_cancellation
from livekit.plugins.turn_detector.multilingual import MultilingualModel
from observability.langfuse import flush_traces, setup_langfuse
from observability.metrics import LatencyTracker
from observability.usage import UsageTracker
from audio.stt import build_stt
from audio.tts import build_tts
from audio.vad import load_vad
from llm.llm import build_llm
from prompts.loader import load_prompt
from mcp_clients.loader import load_mcp_servers_from_metadata
from tools.loader import register_tools


logger = logging.getLogger("agent")

load_dotenv(".env.local")

@register_tools
class Assistant(Agent):
    def __init__(self, user_name: str) -> None:
        base_prompt = load_prompt("wil")
        formatted_instructions = base_prompt.replace("{{user_name}}", user_name)

        super().__init__(
            instructions=formatted_instructions,
        )


server = AgentServer()


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = load_vad()


server.setup_fnc = prewarm


@server.rtc_session()
async def my_agent(ctx: JobContext):
    # Logging setup
    # Add any other context you want in all log entries here
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }

    # Setup Langfuse tracing
    trace_provider = setup_langfuse(
        metadata={
            "langfuse.session.id": ctx.room.name,
            "agent": "assistant",
            "lang": "es",
        }
    )

    # Callback to flush traces when the session ends
    if trace_provider:
        async def flush_langfuse_traces():
            flush_traces(trace_provider)

        ctx.add_shutdown_callback(flush_langfuse_traces)
    
    user_name = "Usuario" 
    active_mcp_servers = []

    # Debug: Log metadata availability from both sources
    room_meta = ctx.room.metadata if ctx.room.metadata else ""
    job_meta = ctx.job.metadata if hasattr(ctx.job, 'metadata') and ctx.job.metadata else ""
    
    <http://logger.info|logger.info>(f"Room metadata available: {ctx.room.metadata is not None}, value: '{room_meta}'")
    <http://logger.info|logger.info>(f"Job metadata available: {hasattr(ctx.job, 'metadata') and ctx.job.metadata is not None}, value: '{job_meta}'")
    <http://logger.info|logger.info>(f"Room name: {ctx.room.name}")

    # Try to get metadata from room first, then fallback to job
    metadata_source = None
    metadata_json = None
    
    try:
        # First try room metadata - check it's not None and not empty string
        if ctx.room.metadata and ctx.room.metadata.strip():
            metadata_source = "room"
            metadata_json = ctx.room.metadata.strip()
            <http://logger.info|logger.info>(f"Using metadata from room: {metadata_json}")
        # Fallback to job metadata if room metadata is not available
        elif hasattr(ctx.job, 'metadata') and ctx.job.metadata and ctx.job.metadata.strip():
            metadata_source = "job"
            metadata_json = ctx.job.metadata.strip()
            <http://logger.info|logger.info>(f"Using metadata from job: {metadata_json}")
        
        if metadata_json:
            <http://logger.info|logger.info>(f"Parsing metadata from {metadata_source}: {metadata_json}")
            meta_data = json.loads(metadata_json)
            <http://logger.info|logger.info>(f"Parsed metadata: {json.dumps(meta_data, indent=2, ensure_ascii=False)}")

            user_name = meta_data.get("user_name", "Usuario")
            <http://logger.info|logger.info>(f"User context: {user_name}")

            active_mcp_servers = load_mcp_servers_from_metadata(metadata_json)
            <http://logger.info|logger.info>(f"Loaded {len(active_mcp_servers)} MCP servers")
        else:
            logger.warning("No metadata found in room or job")
            
    except Exception as e:
        logger.error(f"Error parsing metadata: {e}", exc_info=True)

    # Set up a voice AI pipeline using OpenAI, Cartesia, AssemblyAI, and the LiveKit turn detector
    session = AgentSession(
        stt=build_stt(language="es"),
        llm=build_llm(),
        tts=build_tts(language="es"),

        turn_detection=MultilingualModel(),
        mcp_servers=active_mcp_servers,
        
        vad=ctx.proc.userdata["vad"],
        max_endpointing_delay=1.0,
        min_endpointing_delay=0.3,
        min_interruption_duration=0.2,
        # allow the LLM to generate a response while waiting for the end of turn
        # See more at <https://docs.livekit.io/agents/build/audio/#preemptive-generation>
        preemptive_generation=True,
    )

    latency = LatencyTracker()
    usage = UsageTracker()

    @session.on("metrics_collected")
    def on_metrics(ev: MetricsCollectedEvent):
        m = ev.metrics
        logger.debug(
            f"metrics_collected: type={type(m).__name__}"
        )
        latency.collect(m)
        usage.collect(m)

    @session.on("function_tools_executed")
    def on_tools_executed(event: FunctionToolsExecutedEvent):
        <http://logger.info|logger.info>("🔧 Tools executed")

        for call, output in event.zipped():
            <http://logger.info|logger.info>(f"Tool name: {call.name}")

            # El output suele ser dict / list / str
            try:
                <http://logger.info|logger.info>(
                    "Tool output:\n%s",
                    json.dumps(output.output, indent=2, ensure_ascii=False)
                )
            except Exception:
                <http://logger.info|logger.info>(f"Tool output (raw): {output.output}")

    
    async def log_session_metrics():
        latency_summary = latency.summary()

        <http://logger.info|logger.info>(
            f"[room={ctx.room.name}] latency_summary={latency_summary}"
        )

        <http://logger.info|logger.info>(
            f"[room={ctx.room.name}] usage={usage.summary()}"
        )

    ctx.add_shutdown_callback(log_session_metrics)

    ctx.log_context_fields |= {
        "agent": "assistant",
        "lang": "es",
    }
    # Start the session, which initializes the voice pipeline and warms up the models
    await session.start(
        agent=Assistant(user_name=user_name),
        room=ctx.room,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
                if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                else noise_cancellation.BVC(),
            ),
        ),
    )

    # Join the room and connect to the user
    await ctx.connect()
    
    # Check metadata again after connecting (in case it's only available after connection)
    <http://logger.info|logger.info>(f"After connect - Room metadata: '{ctx.room.metadata if ctx.room.metadata else ''}'")
    if hasattr(ctx.job, 'metadata'):
        <http://logger.info|logger.info>(f"After connect - Job metadata: '{ctx.job.metadata if ctx.job.metadata else ''}'")
    
    # Debug: Log all available attributes of ctx.job
    <http://logger.info|logger.info>(f"Job attributes: {[attr for attr in dir(ctx.job) if not attr.startswith('_')]}")
    if hasattr(ctx.job, 'metadata'):
        <http://logger.info|logger.info>(f"Job metadata type: {type(ctx.job.metadata)}, value: '{ctx.job.metadata}'")


if __name__ == "__main__":
    cli.run_app(
        server
    )

Thanks — your logs actually show the root cause: the room metadata is empty when the job is assigned, but populated after you connect. Your agent is already doing the right checks; the timing is the issue. What’s happening (concise) • Room metadata can be absent or empty at assignment time. The full room state (including metadata) is guaranteed after the agent connects to the room. • Your logs confirm this: initial logs show room metadata value '', then after

await ctx.connect()

you log the metadata JSON. What to do (concise) • Read and parse

ctx.room.metadata

only after

await ctx.connect()

(not before). Move metadata parsing to after connect, or re-check it after connect. • Keep a single source: prefer room metadata for user info shared across agents. Suggested minimal change 1) Connect first, then parse:

await ctx.connect()

room_meta = (ctx.room.metadata or "").strip()
if room_meta:
    meta_data = json.loads(room_meta)
    user_name = meta_data.get("user_name", "Usuario")
    active_mcp_servers = load_mcp_servers_from_metadata(room_meta)
else:
    logger.warning("Room metadata empty after connect")

2) Optional: if you still want to allow job metadata fallback, check job metadata after connect as well. Debug tips • Log

ctx.room.metadata

immediately after

await ctx.connect()

(you already do that). • Ensure the room/token actually contains the metadata on creation: verify the backend that creates the token/room sets

RoomConfiguration.metadata

to the JSON string. If you want, paste the code you use to generate the join token or create the room and I’ll point out exactly where to set the JSON metadata.

in terms of latency, is it better to read metadata from the room, from the job, or is it the same?

Short answer: latency is effectively the same — reading either

ctx.room.metadata

or

ctx.job.metadata

is just a local in-memory access once the data is available. The real difference is availability and timing, not read latency. Key points: • Read cost: both are simple string reads and JSON parsing; negligible latency compared to network/processing. • Availability/timing: ◦

ctx.job.metadata

is available immediately with the job if you dispatch the agent with metadata. ◦

ctx.room.metadata

may not be populated until after the agent connects; you should read it after

await ctx.connect()

. • Reliability for shared data: use room metadata when multiple agents/participants must see the same user info. Use job metadata for per-agent, per-dispatch config. • Best practice: ◦ For shared user info (your case) store it in room metadata and parse it after connect. ◦ If you need guaranteed immediate access at assignment time, dispatch the agent with job metadata instead. ◦ Keep metadata small; for secrets/tokens prefer references and fetch server-side. If you want, tell me whether you need metadata at assignment time (before connect) or only after the agent starts handling audio — I’ll recommend exact code changes.