when livekit is exporting otel traces, is there a way to export them as json spans instead of protobufs?

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

I'm using python

Thanks — you're asking about exporting LiveKit OpenTelemetry traces as JSON spans from the Python SDK. LiveKit uses the standard OpenTelemetry pipeline in the Python agents, and the OTLP exporter supports JSON over HTTP. To export JSON spans, configure the OTLP HTTP span exporter in your Python process (instead of the default gRPC/protobuf exporter). Example:

from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

exporter = OTLPSpanExporter(
    endpoint="<https://your-collector-endpoint/v1/traces>",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
)

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
# set global provider if needed
# from opentelemetry import trace
# trace.set_tracer_provider(tracer_provider)

Notes: • Ensure your collector/observability backend accepts OTLP/HTTP JSON at the endpoint you provide. • LiveKit generates spans in protobuf internally, but exporting as JSON is handled by the OTLP HTTP exporter you configure in Python. Would you like the exact snippet adapted to your collector URL or help switching from an existing gRPC exporter? Sources: Examples | livekit/livekit_composite | DeepWiki | Setting custom user_id and session_id in Langfuse traces

In the above code, where there is no configuration to export spans as json. Can you give the full example where I can export them as jsons

Thanks — I hear you want an explicit, runnable Python example that forces OTLP/HTTP to send JSON-encoded spans (not protobuf). Below is a complete example that: • Uses the OTLP HTTP exporter (this sends JSON by default for the HTTP/JSON OTLP path), • Shows environment variable options and explicit exporter args, • Sets up a tracer, creates a test span, and flushes.

# requirements:
# pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp

import os
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.trace import Tracer

# --- Option A: Set via environment variables (OTLP/HTTP JSON) ---
# OTEL_EXPORTER_OTLP_ENDPOINT should include the /v1/traces path for many collectors.
# Setting OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf vs http/proto is not standardized across all SDKs,
# but using the HTTP exporter class below ensures HTTP/JSON is used.
os.environ.setdefault("OTEL_EXPORTER_OTLP_ENDPOINT", "<https://your-collector.example.com/v1/traces>")
# If your collector needs auth headers:
os.environ.setdefault("OTEL_EXPORTER_OTLP_HEADERS", "authorization=Bearer YOUR_TOKEN")

# --- Option B: Explicit exporter construction (preferred to guarantee HTTP exporter) ---
exporter = OTLPSpanExporter(
    endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "<https://your-collector.example.com/v1/traces>"),
    headers={"authorization": "Bearer YOUR_TOKEN"},  # or read from env
    # The HTTP OTLP exporter class used here sends JSON-serialized OTLP over HTTP.
)

resource = Resource.create({"service.name": "livekit-python-example"})

provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(exporter)
provider.add_span_processor(processor)

# Make this provider global
trace.set_tracer_provider(provider)
tracer: Tracer = trace.get_tracer(__name__)

# Create a test span
with tracer.start_as_current_span("example-span") as span:
    span.set_attribute("example.attr", "value")
    # do work...

# Ensure exporter flushes before exit
provider.shutdown()

Notes: • Using

OTLPSpanExporter

from

opentelemetry.exporter.otlp.proto.http.trace_exporter

ensures the HTTP exporter is used; that exporter sends OTLP/JSON over HTTP (collector endpoint

/v1/traces

). • Some collectors accept both protobuf/gRPC and JSON/HTTP on different endpoints; confirm your collector URL supports OTLP/HTTP JSON at /v1/traces. • If you previously used the gRPC exporter or a protobuf-based exporter class, switch to this HTTP exporter class to get JSON encoding. If you tell me your collector type (e.g., OpenTelemetry Collector, Honeycomb, Datadog) and the endpoint you must use, I can adapt this snippet to match any required header/auth or endpoint differences.

I'm currently using langfuse otel exporter, now I want to switch to my custom exporter which only accepts json and not protobuf. tell me how do I switch?

You're switching from the Langfuse OTLP exporter to a custom exporter that only accepts JSON spans. I'll explain two clear options and give a minimal, ready-to-plug Python example you can wire into your LiveKit Python agent. Which of these fits you best? • Replace the OTLP exporter with a custom SpanExporter that serializes spans to OTLP/JSON and POSTs them to your endpoint. • Run the OTLP/HTTP JSON exporter but route it through a small proxy that converts protobuf -> JSON (useful if you want to reuse existing exporter plumbing). Below is a simple custom

SpanExporter

implementation that serializes spans to JSON and sends them to your endpoint. Wire it into OpenTelemetry the same way you currently wire Langfuse. Code (requires opentelemetry-sdk and requests):

# pip install opentelemetry-sdk requests

import json
import time
import requests
from typing import Sequence
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult, BatchSpanProcessor
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider

class JsonHttpSpanExporter(SpanExporter):
    def __init__(self, endpoint: str, headers: dict | None = None, timeout: int = 10):
        self.endpoint = endpoint
        self.headers = headers or {"Content-Type": "application/json"}
        self.timeout = timeout

    def _span_to_dict(self, span: ReadableSpan) -> dict:
        return {
            "name": span.name,
            "context": {
                "trace_id": format(span.context.trace_id, "032x"),
                "span_id": format(span.context.span_id, "016x"),
            },
            "start_time": int(span.start_time / 1e6),  # ms
            "end_time": int(span.end_time / 1e6),      # ms
            "attributes": {k: v for k, v in span.attributes.items()},
            "status": {"status_code": int(span.status.status_code)},
        }

    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        try:
            payload = {"spans": [self._span_to_dict(s) for s in spans]}
            resp = <http://requests.post|requests.post>(self.endpoint, data=json.dumps(payload), headers=self.headers, timeout=self.timeout)
            if resp.status_code >= 200 and resp.status_code < 300:
                return SpanExportResult.SUCCESS
            # treat non-2xx as retryable if 429/5xx
            if resp.status_code == 429 or 500 <= resp.status_code < 600:
                return SpanExportResult.FAILURE  # SDK may retry based on processor
            return SpanExportResult.SUCCESS
        except Exception:
            return SpanExportResult.FAILURE

    def shutdown(self) -> None:
        pass

# Wire into OpenTelemetry
endpoint = "<https://your-json-collector.example.com/v1/spans>"
headers = {"Authorization": "Bearer YOUR_TOKEN", "Content-Type": "application/json"}

exporter = JsonHttpSpanExporter(endpoint=endpoint, headers=headers)
provider = TracerProvider(resource=Resource.create({"service.name": "livekit-python"}))
processor = BatchSpanProcessor(exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Example usage
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("test-span") as span:
    span.set_attribute("foo", "bar")

# ensure flush on exit
provider.shutdown()

Notes and suggestions: • The example uses a simple JSON schema (list under "spans"). Adjust field names/format (timestamps, attributes, status) to match your custom collector's expected schema. • If you need OTLP/JSON compatibility (the OTLP HTTP JSON wire-format), you can adapt

_span_to_dict

to produce OTLP JSON shape. The OpenTelemetry project provides OTLP proto -> JSON mapping — ask if you want that exact mapping. • Use

BatchSpanProcessor

as shown for production; it buffers and sends in batches like standard exporters. • If you previously configured Langfuse with environment variables, replace that wiring with this exporter and ensure you call

trace.set_tracer_provider(provider)

once at startup. If you share your collector's expected JSON schema or an example payload, I’ll adapt the exporter to match it exactly.

great, this is good. Is there a way that I can set a global ID for each of these spans such as a sessionId or callId?

Great — you can add a global ID to every span in three common ways. I’ll show minimal examples for each and the recommended approach. Options • Resource-level attribute (recommended when ID is truly global to the process, e.g., service-wide session). • SpanProcessor that injects the ID into every span on export (recommended when you want to attach before export, e.g., call-scoped IDs added by instrumentation layer). • Set attribute at span creation (recommended when you have the ID in your request-handling context). 1) Resource-level attribute (simple, set once at provider start)

from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider

resource = Resource.create({"service.name": "livekit-python", "session_id": "GLOBAL_SESSION_123"})
provider = TracerProvider(resource=resource)
# add processor/exporter as before...

2) SpanProcessor that injects an ID into every span on end (good if you can determine the ID at runtime or update it per-call)

from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.sdk.trace.export import SpanProcessor

class InjectIdSpanProcessor(SpanProcessor):
    def __init__(self, id_getter):
        # id_getter: callable that returns current id (e.g., from thread-local or context)
        self.id_getter = id_getter

    def on_start(self, span, parent_context):
        pass

    def on_end(self, span: ReadableSpan):
        current_id = self.id_getter()
        if current_id:
            span.attributes["session_id"] = current_id

# Example id_getter using contextvar
import contextvars
current_session = contextvars.ContextVar("current_session", default=None)
def id_getter():
    return current_session.get()

# register processor
provider.add_span_processor(InjectIdSpanProcessor(id_getter))

3) Set attribute when creating spans (best when handling per-request/call context explicitly)

from opentelemetry import trace

tracer = trace.get_tracer(__name__)
# set context-local session id before work (example using contextvar)
current_session.set("CALL_456")
with tracer.start_as_current_span("call-span") as span:
    span.set_attribute("session_id", "CALL_456")
    # work...

Which to use • If the ID is global for the whole process: Resource attribute. • If the ID changes per request/call and you can access it via a context variable: SpanProcessor or set on span creation. SpanProcessor is convenient when you want instrumentation-agnostic injection. • If you want the exporter payload to include the ID regardless of instrumentation, the SpanProcessor ensures it’s present at export time. If you want, tell me: • whether the ID is per-process, per-call, or per-user, and • how you currently propagate request context (contextvars, thread locals, headers), and I’ll give the exact code wired into your JsonHttpSpanExporter example.

my ID is per call

Thanks — you said the ID is per-call. The cleanest approach is to inject the callId into every span at runtime using a SpanProcessor that reads the callId from a contextvar (set when handling the call). That keeps instrumentation unchanged and guarantees the callId is present on every span at export time. Below is a minimal, ready-to-plug example that: • Uses your custom JSON exporter from earlier • Uses a contextvar to store the per-call id • Adds an InjectCallIdSpanProcessor that writes

call_id

to each span on end

# pip install opentelemetry-sdk requests

import json
import requests
import uuid
import contextvars
from typing import Sequence
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider, ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult, BatchSpanProcessor, SpanProcessor

# ---- contextvar to hold current call id ----
current_call_id: contextvars.ContextVar[str | None] = contextvars.ContextVar("current_call_id", default=None)

# ---- JSON HTTP exporter (basic) ----
class JsonHttpSpanExporter(SpanExporter):
    def __init__(self, endpoint: str, headers: dict | None = None, timeout: int = 10):
        self.endpoint = endpoint
        self.headers = headers or {"Content-Type": "application/json"}
        self.timeout = timeout

    def _span_to_dict(self, span: ReadableSpan) -> dict:
        return {
            "name": span.name,
            "trace_id": format(span.context.trace_id, "032x"),
            "span_id": format(span.context.span_id, "016x"),
            "start_time_ms": int(span.start_time / 1e6),
            "end_time_ms": int(span.end_time / 1e6),
            "attributes": dict(span.attributes) if span.attributes else {},
            "status": int(span.status.status_code),
            "events": [
                {"name": ev.name, "timestamp_ms": int(ev.timestamp / 1e6), "attributes": dict(ev.attributes or {})}
                for ev in span.events
            ],
        }

    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        try:
            payload = {"spans": [self._span_to_dict(s) for s in spans]}
            resp = <http://requests.post|requests.post>(self.endpoint, json=payload, headers=self.headers, timeout=self.timeout)
            if 200 <= resp.status_code < 300:
                return SpanExportResult.SUCCESS
            if resp.status_code == 429 or 500 <= resp.status_code < 600:
                return SpanExportResult.FAILURE
            return SpanExportResult.SUCCESS
        except Exception:
            return SpanExportResult.FAILURE

    def shutdown(self) -> None:
        pass

# ---- SpanProcessor that injects call_id into span attributes on end ----
class InjectCallIdSpanProcessor(SpanProcessor):
    def __init__(self, attr_name: str = "call_id"):
        self.attr_name = attr_name

    def on_start(self, span, parent_context):
        return None

    def on_end(self, span: ReadableSpan):
        call_id = current_call_id.get()
        if call_id:
            # update span attributes in-place
            span.attributes[self.attr_name] = call_id

# ---- Setup function to create tracer provider per-process (call-specific call_id set per call) ----
def setup_tracing(json_endpoint: str, service_name: str = "livekit-python"):
    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)

    exporter = JsonHttpSpanExporter(endpoint=json_endpoint, headers={"Content-Type": "application/json"})
    # Use BatchSpanProcessor in production
    provider.add_span_processor(InjectCallIdSpanProcessor())
    provider.add_span_processor(BatchSpanProcessor(exporter))

    trace.set_tracer_provider(provider)
    return provider

# ---- Example: how to use within a per-call context (e.g., inside your LiveKit call handler) ----
from contextlib import contextmanager
from opentelemetry import trace as ot_trace

@contextmanager
def call_context(call_id: str | None = None):
    """Context manager to set the call_id for the duration of a call."""
    token = None
    try:
        if call_id is None:
            call_id = f"call-{uuid.uuid4().hex[:8]}"
        token = current_call_id.set(call_id)
        yield call_id
    finally:
        if token is not None:
            current_call_id.reset(token)

# ---- Minimal usage example ----
if __name__ == "__main__":
    # configure once at process start
    provider = setup_tracing(json_endpoint="<https://your-json-collector.example.com/v1/spans>")

    tracer = ot_trace.get_tracer(__name__)

    # Simulate handling a call — set call_id for this call's lifetime
    with call_context("CALL_12345"):
        # all spans created here will get attribute "call_id": "CALL_12345"
        with tracer.start_as_current_span("incoming-request") as span:
            span.set_attribute("user", "alice")
            with tracer.start_as_current_span("do-work"):
                pass

    # flush before exit
    provider.shutdown()

Notes • Set the contextvar at the start of each call (e.g., in your LiveKit handler where you can derive the call id from ctx.room, headers, or job context). • The InjectCallIdSpanProcessor writes

call_id

to every ended span, so it will appear in your exported JSON regardless of how spans were created. • If you prefer a different attribute name (e.g.,

session_id

or

call.session_id

), change the

InjectCallIdSpanProcessor(attr_name=...)

accordingly. • Keep BatchSpanProcessor for production to buffer and batch exports. If you tell me where you generate call ids in your LiveKit handler (e.g.,

ctx.room.name

, job context field), I’ll show the exact place to call

call_context(...)

so it integrates with your agent code. Sources: Job lifecycle | LiveKit docs | Context Variables | LiveKit docs