First-class OpenTelemetry support when running ADK agents outside Google Cloud

[rwang1987]

Is your feature request related to a problem? Please describe.
In our company, we are piloting the Agent Development Kit to run AI agents entirely on-prem (inside an air-gapped Google Distributed Cloud rack).
While opentelemetry-api and opentelemetry-sdk are part of ADK’s dependencies, the current documentation only explains how telemetry is emitted when agents run on Vertex AI Agent Engine. There is no guidance or helper for configuring OTLP exporters when we deploy agents on our own Kubernetes cluster.

Describe the solution you’d like

• A simple telemetry config block (or CLI flags) that lets us specify:
  • OTLP endpoint / headers
  • Service name / version
  • Sampling strategy
• Optional automatic instrumentation for:
  • LLM requests (trace_call_llm)
  • Tool invocations (trace_tool_call)
  • Agent-to-agent (A2A) messaging
• Docs + example showing how to push traces to an on-prem collector (Jaeger, Tempo, GCP Ops Agent in hybrid mode, etc.).

Describe alternatives you’ve considered

• Manually initialising the OTEL SDK in our main.py and decorating each callback.
  Works, but duplicates boilerplate and misses deeper internals (e.g. planner traces).
• Exporting logs only – doesn’t give us span correlation across micro-services.

Additional context
We already have an OpenTelemetry collector running inside our data-centre; exposing ADK spans would let us view agent reasoning latency alongside the rest of our CI/CD traces.

Thanks for considering!

[aabmass]

@rwang1987 have you seen #320? cc @michaelsafyan

[rwang1987]

Thank you @aabmass and @michaelsafyan will you also help to update the ADK documentation to provide instructions of OTel export part? That will be super cool! If that's the case, I guess we can close this ticket?

[michaelsafyan]

If you don't mind, it would be helpful for tracking and for prioritization/acceptance discussions with other contributors/maintainers to leave this open until the request is fully resolved.

But, yes, we would also like to ensure that documentation is updated and contains appropriate examples.

Realistically, the pull request that Aaron linked to will probably need to be broken down into smaller pull requests to make it in (as well as for simplifying incremental testing of its various pieces). Is there anything you can share with us regarding the level of urgency of this request? (Also helpful as a prioritization signal for us).

[rwang1987]

@michaelsafyan Yes, for sure, I would love to keep this ticket open!

[Jainish-S]

@rwang1987 Not sure exactly about the issue but I export traces like this, if its any helpful to you

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

LANGFUSE_PUBLIC_KEY = "pk-lf-8d838188-6390-4cb5-a30f-0a58c3d30752"
LANGFUSE_SECRET_KEY = "sk-lf-30df3ce3-472b-4018-a1fb-ae6866ce65c2"
LANGFUSE_AUTH = base64.b64encode(
    f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    "http://localhost:3000/api/public/otel"  # Original Local dev for Langfuse or mention your endpoint 
)
# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://us.cloud.langfuse.com/api/public/otel" # US data region
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

provider = TracerProvider(resource=Resource.create({"service.name": "langfuse"}))
exporter = OTLPSpanExporter()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("bcc_agents")

from fastapi import FastAPI, Request, HTTPException
import logging
import json
import os
from datetime import datetime
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
    ExportTraceServiceRequest,
)
from google.protobuf.json_format import MessageToJson

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

# Create data directory if it doesn't exist
DATA_DIR = "data"
os.makedirs(DATA_DIR, exist_ok=True)

# Global counter for incremental request IDs
request_counter = 0


def save_payload_to_file(payload_data, content_type, request_id=None):
    """Save payload data to a JSON file in the data directory."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    if request_id:
        filename = f"payload_{timestamp}_{request_id}.json"
    else:
        filename = f"payload_{timestamp}.json"

    filepath = os.path.join(DATA_DIR, filename)

    try:
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(payload_data, f, indent=2, ensure_ascii=False)
        logger.info(f"Payload saved to: {filepath}")
        return filepath
    except Exception as e:
        logger.error(f"Error saving payload to file: {e}", exc_info=True)
        return None


@app.post("/v1/traces")
async def receive_traces(request: Request):
    global request_counter
    content_type = request.headers.get("content-type")
    logger.info(f"Received request to /v1/traces with Content-Type: {content_type}")

    # Generate an incremental numeric request ID
    request_counter += 1
    request_id = request_counter

    try:
        raw_payload = await request.body()
        logger.info(f"Received raw payload (length: {len(raw_payload)} bytes)")

        payload_to_save = None

        if content_type == "application/x-protobuf":
            try:
                trace_request = ExportTraceServiceRequest()
                trace_request.ParseFromString(raw_payload)
                # Pretty print the JSON output
                json_payload = MessageToJson(trace_request, indent=2)
                logger.info(
                    f"Received OTLP Protobuf payload (converted to JSON):\n{json_payload}"
                )
                # Parse the JSON string to a dict for saving
                payload_to_save = json.loads(json_payload)
            except Exception as e:
                logger.error(
                    f"Error parsing/converting OTLP Protobuf payload: {e}",
                    exc_info=True,
                )
                logger.info(
                    f"Raw Protobuf payload (hex) for failed conversion: {raw_payload.hex()}"
                )
                # Save the raw hex data if parsing fails
                payload_to_save = {"raw_hex": raw_payload.hex(), "error": str(e)}

        elif content_type == "application/json":
            # If OTLP exporter was configured for JSON
            try:
                payload_json = json.loads(raw_payload.decode("utf-8"))
                logger.info(
                    f"Received OTLP JSON payload:\n{json.dumps(payload_json, indent=2)}"
                )
                payload_to_save = payload_json
            except Exception as e:
                logger.error(f"Error decoding OTLP JSON payload: {e}", exc_info=True)
                logger.info(f"Raw payload (hex) for failed JSON: {raw_payload.hex()}")
                # Save the raw hex data if parsing fails
                payload_to_save = {"raw_hex": raw_payload.hex(), "error": str(e)}
        else:
            # Fallback for unexpected content types
            logger.warning(
                f"Unexpected Content-Type: {content_type}. Attempting to log raw payload as hex."
            )
            logger.info(f"Raw payload (hex): {raw_payload.hex()}")
            # Save the raw hex data for unexpected content types
            payload_to_save = {"raw_hex": raw_payload.hex()}

        # Save the payload to file
        if payload_to_save:
            save_payload_to_file(payload_to_save, content_type, request_id)

        return {"message": "Traces received successfully"}
    except Exception as e:
        logger.error(f"Error processing /v1/traces request: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail="Internal server error")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=1111)

[rwang1987]

@Jainish-S thank you for providing the sample code, but have you figure out a way to enable Opentelemetry to work with adk commands like adk web, adk run, adk api_server? Just using the root agent, not with any customized runner.

[Jainish-S]

@rwang1987 Currently we don't have any options to pass it like an argument in the adk web or any cli command. But we can do this

File: google/adk/cli/fast_api.py

def get_fast_api_app(
    *,
    agent_dir: str,
    session_db_url: str = "",
    allow_origins: Optional[list[str]] = None,
    web: bool,
    trace_to_cloud: bool = False,
    lifespan: Optional[Lifespan[FastAPI]] = None,
) -> FastAPI:
  # InMemory tracing dict.
  trace_dict: dict[str, Any] = {}

  # Set up tracing in the FastAPI server.
  provider = TracerProvider()
  provider.add_span_processor(
      export.SimpleSpanProcessor(ApiServerSpanExporter(trace_dict))
  )
  if trace_to_cloud:
    envs.load_dotenv_for_agent("", agent_dir)
    if project_id := os.environ.get("GOOGLE_CLOUD_PROJECT", None):
      processor = export.BatchSpanProcessor(
          CloudTraceSpanExporter(project_id=project_id)
      )
      provider.add_span_processor(processor)
    else:
      logging.warning(
          "GOOGLE_CLOUD_PROJECT environment variable is not set. Tracing will"
          " not be enabled."
      )

  trace.set_tracer_provider(provider)

One thing we can do is set the tracer before initializing the app like this

import os
import uvicorn

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

LANGFUSE_PUBLIC_KEY = "pk-lf-8d838188-6390-4cb5-a30f-0a58c3d30752"
LANGFUSE_SECRET_KEY = "sk-lf-30df3ce3-472b-4018-a1fb-ae6866ce65c2"
LANGFUSE_AUTH = base64.b64encode(
    f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    "http://localhost:3000/api/public/otel"  # Original Local dev for Langfuse or mention your endpoint 
)
# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://us.cloud.langfuse.com/api/public/otel" # US data region
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

provider = TracerProvider(resource=Resource.create({"service.name": "langfuse"}))
exporter = OTLPSpanExporter()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agents")


app = get_fast_api_app(
    agents_dir=".", session_db_url="", allow_origins=["*"], web=True, trace_to_cloud=False
)

config = uvicorn.Config(
    app,
    host="0.0.0.0",
    port=8090,
)

server = uvicorn.Server(config)
server.run()

[svelezdevilla]

@Jainish-S

@rwang1987 Currently we don't have any options to pass it like an argument in the adk web or any cli command. But we can do this

File: google/adk/cli/fast_api.py

def get_fast_api_app(
*,
agent_dir: str,
session_db_url: str = "",
allow_origins: Optional[list[str]] = None,
web: bool,
trace_to_cloud: bool = False,
lifespan: Optional[Lifespan[FastAPI]] = None,
) -> FastAPI:

InMemory tracing dict.

trace_dict: dict[str, Any] = {}

Set up tracing in the FastAPI server.

provider = TracerProvider()
provider.add_span_processor(
export.SimpleSpanProcessor(ApiServerSpanExporter(trace_dict))
)
if trace_to_cloud:
envs.load_dotenv_for_agent("", agent_dir)
if project_id := os.environ.get("GOOGLE_CLOUD_PROJECT", None):
processor = export.BatchSpanProcessor(
CloudTraceSpanExporter(project_id=project_id)
)
provider.add_span_processor(processor)
else:
logging.warning(
"GOOGLE_CLOUD_PROJECT environment variable is not set. Tracing will"
" not be enabled."
)

trace.set_tracer_provider(provider)
One thing we can do is set the tracer before initializing the app like this

import os
import uvicorn

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

LANGFUSE_PUBLIC_KEY = "pk-lf-8d838188-6390-4cb5-a30f-0a58c3d30752"
LANGFUSE_SECRET_KEY = "sk-lf-30df3ce3-472b-4018-a1fb-ae6866ce65c2"
LANGFUSE_AUTH = base64.b64encode(
    f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    "http://localhost:3000/api/public/otel"  # Original Local dev for Langfuse or mention your endpoint 
)
# os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "https://us.cloud.langfuse.com/api/public/otel" # US data region
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

provider = TracerProvider(resource=Resource.create({"service.name": "langfuse"}))
exporter = OTLPSpanExporter()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agents")


app = get_fast_api_app(
    agents_dir=".", session_db_url="", allow_origins=["*"], web=True, trace_to_cloud=False
)

config = uvicorn.Config(
    app,
    host="0.0.0.0",
    port=8090,
)

server = uvicorn.Server(config)
server.run()

Hello, I tried your approach, and while the trace appears in langfuse, a few things don't work as expected.

1. When executing that script, I get: Overriding of current TracerProvider is not allowed.
2. Missing user_id and session_id in Langfuse.
3. Input / Output fields appear blank.

Currently using langfuse==3.1.2 and google-adk==1.5.0.

Any idea how this could be fixed?

Thank you very much!