Tracing LangGraph Agents on Vertex AI Agent Engine With OpenTelemetry

Deploy a LangGraph agent to Google Cloud's Agent Engine, open Cloud Trace and you'll find an empty screen. Agent Engine is a managed runtime that handles autoscaling, checkpointing and serverless execution after you package the agent and call a create API. What it doesn't do is instrument your agent. There's no span tree, no LLM call latency, nothing to drill into when a user asks why a request took 30 seconds or why an answer was wrong.

The fix is a repeatable setup: start OpenTelemetry when the agent boots, wrap important functions in custom spans and ship everything to Cloud Trace, without asking the author of the agent YAML to change anything.

Turning tracing on at startup

Inside a custom Agent Engine environment there's no web server lifecycle to hook into. The process starts cold from a serialized runnable, so initialization code has to do the setup itself, and the first thing it should call is a function that takes a service name:

set_opentelemetry("my-agent-service")

That one call wires up all three OpenTelemetry signals. Traces go to Cloud Trace over OTLP/gRPC, logs to Cloud Logging and metrics to Cloud Monitoring.

Two design choices matter. The function is idempotent, so calling it from several code paths won't rebuild the provider stack. And it's built to fail softly. If credentials or an IAM role are missing, it writes a warning and the agent keeps serving traffic, because a telemetry misconfiguration must never take a production agent down. Credentials come from Application Default Credentials, which means nothing extra is needed when the runtime already has a service account attached.

What the full setup does

Each signal gets its own error handling, so a failure in metrics can't stop traces from working:

def set_opentelemetry(service_name: str) -> None:
    global _initialized
    if _initialized:
        return
    _initialized = True
    os.environ["OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED"] = "true"
    os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = "true"
    os.environ["OTEL_SERVICE_NAME"] = service_name
    gcp_creds, project_id = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    service_resource = Resource.create({
        SERVICE_NAME: service_name,
        "gcp.project_id": project_id,
    })
    auth_plugin = AuthMetadataPlugin(credentials=gcp_creds, request=Request())
    grpc_channel_creds = grpc.composite_channel_credentials(
        grpc.ssl_channel_credentials(),
        grpc.metadata_call_credentials(auth_plugin),
    )
    try:
        tracer_provider = TracerProvider(resource=service_resource)
        tracer_provider.add_span_processor(CredentialRedactingSpanProcessor())
        tracer_provider.add_span_processor(
            BatchSpanProcessor(
                OTLPSpanExporter(
                    credentials=grpc_channel_creds,
                    endpoint="https://telemetry.googleapis.com:443/v1/traces",
                )
            )
        )
        trace.set_tracer_provider(tracer_provider)
    except Exception as exc:
        logger.warning("Trace export setup failed: %s", exc)
        trace.set_tracer_provider(TracerProvider(resource=service_resource))
    try:
        log_provider = LoggerProvider(resource=service_resource)
        log_provider.add_log_record_processor(BatchLogRecordProcessor(CloudLoggingExporter()))
        logs.set_logger_provider(log_provider)
        event_log_provider = EventLoggerProvider(log_provider)
        events.set_event_logger_provider(event_log_provider)
    except Exception as exc:
        logger.warning("Cloud Logging export setup failed: %s", exc)
    try:
        metrics_exporter = CloudMonitoringMetricsExporter()
        metric_reader = PeriodicExportingMetricReader(metrics_exporter)
        meter_provider = MeterProvider(metric_readers=[metric_reader], resource=service_resource)
        metrics.set_meter_provider(meter_provider)
    except Exception as exc:
        logger.warning("Cloud Monitoring metrics setup failed: %s", exc)
    for cls, name in (
        (HTTPXClientInstrumentor, "httpx"),
        (SQLite3Instrumentor, "sqlite3"),
        (VertexAIInstrumentor, "vertexai"),
    ):
        try:
            cls().instrument()
        except Exception as exc:
            logger.warning("Instrumentation skipped for %s: %s", name, exc)

The traces go to telemetry.googleapis.com and not to a standard collector. Google's Telemetry API is the recommended OTLP ingestion endpoint for Cloud Trace. It accepts plain OTLP/gRPC, authenticates with Google OAuth2 and writes straight to Cloud Trace. With no collector sidecar to run, it suits Agent Engine's serverless model.

Keeping secrets out of spans

LangGraph's OpenInference instrumentation serializes the whole agent state dictionary into span attributes, and that state may hold runtime credentials such as API keys or tokens for enterprise tools. A custom span processor scrubs them before any span leaves the process:

class CredentialRedactingSpanProcessor(SpanProcessor):
    _REDACT_KEYS = frozenset({"_runtime_credentials"})

    def on_end(self, span) -> None:
        attrs = getattr(span, "attributes", None)
        if not attrs:
            return
        for key, val in attrs.items():
            if not isinstance(val, str):
                continue
            if not any(f'"{k}"' in val for k in self._REDACT_KEYS):
                continue
            try:
                data = json.loads(val)
                if _redact_dict_recursive(data):
                    attrs[key] = json.dumps(data)
            except Exception:
                for k in self._REDACT_KEYS:
                    val = re.sub(
                        rf'"{re.escape(k)}"\s*:\s*(?:\{{[^{{}}]*\}}|"[^"]*")',
                        f'"{k}": "[REDACTED]"',
                        val,
                    )
                attrs[key] = val

Register it ahead of the BatchSpanProcessor. That ordering guarantees credentials are never queued for export.

Custom spans with a decorator

Auto-instrumentation produces HTTP and Vertex AI spans, but it can't see your business logic. Questions like how long a deployment took, which agent needed 45 seconds or whether a query succeeded need custom spans. A do_trace decorator covers both sync and async functions:

def do_trace(fn:callable | None = None, *, span_name: str | None = None) -> callable:
    def decorator(f:callable) -> callable:
        @functools.wraps(f)
        async def _async_wrap(*args, **kwargs):
            tracer = trace.get_tracer(f.__module__)
            active_span_name = span_name or _build_span_name(f, *args, **kwargs)
            with tracer.start_as_current_span(active_span_name) as current_span:
                _set_span_attributes(current_span, f, *args, **kwargs)
                try:
                    return_value = await f(*args, **kwargs)
                    current_span.set_attribute("execution.success", True)
                    current_span.set_status(trace.Status(trace.StatusCode.OK))
                    _set_span_response(current_span, return_value)
                    return return_value
                except Exception as exc:
                    current_span.set_attribute("execution.success", False)
                    current_span.record_exception(exc)
                    current_span.set_status(trace.Status(trace.StatusCode.ERROR))
                    raise

        @functools.wraps(f)
        def _sync_wrap(*args, **kwargs):
            tracer = trace.get_tracer(f.__module__)
            active_span_name = span_name or _build_span_name(f, *args, **kwargs)
            with tracer.start_as_current_span(active_span_name) as current_span:
                _set_span_attributes(current_span, f, *args, **kwargs)
                try:
                    return_value = f(*args, **kwargs)
                    current_span.set_attribute("execution.success", True)
                    current_span.set_status(trace.Status(trace.StatusCode.OK))
                    _set_span_response(current_span, return_value)
                    return return_value
                except Exception as exc:
                    current_span.set_attribute("execution.success", False)
                    current_span.record_exception(exc)
                    current_span.set_status(trace.Status(trace.StatusCode.ERROR))
                    raise

        if inspect.iscoroutinefunction(f):
            return _async_wrap
        return _sync_wrap
    if fn is None:
        return decorator
    return decorator(fn)

When no span name is passed, the decorator pulls agent_id, resource_id and version from the function arguments and assembles a meaningful name:

def _build_span_name(func, *args, **kwargs) -> str:
    bound = _bind_partial_safe(func, *args, **kwargs)
    parts = []
    agent_id = bound.arguments.get("agent_id")
    resource_id = bound.arguments.get("resource_id")
    version = bound.arguments.get("version")
    if agent_id:
        parts.append(str(agent_id))
    elif resource_id:
        parts.append(str(resource_id))
    if version:
        parts.append(str(version))
    parts.append(func.__name__)
    return ".".join(parts) if parts else f"LanggraphAgent.{func.__name__}"

The payoff is a span called research-agent.v2.invoke instead of a bare invoke, so in Cloud Trace it's obvious which agent and version is slow. The decorator also records the query input and the response on the span, truncated to safe lengths:

def _set_span_attributes(span, func, *args, **kwargs) -> None:
    bound = _bind_partial_safe(func, *args, **kwargs)
    span.set_attribute("code.function", func.__name__)
    span.set_attribute("code.module", func.__module__)
    for attr in ("agent_id", "resource_id", "version"):
        val = bound.arguments.get(attr)
        if val:
            span.set_attribute(f"execution.{attr}", str(val))
    request = bound.arguments.get("request")
    if request is not None:
        raw_input = getattr(request, "input", None)
        if raw_input is not None:
            span.set_attribute("query.input", str(raw_input)[:1000])
        session_id = getattr(request, "session_id", None)
        if session_id:
            span.set_attribute("query.session_id", str(session_id))

Applying spans at each layer

At the API service layer, a named span marks the entry point:

@do_trace(span_name="api.install_agent")
async def install_agent(agent_id: str, request: DeploymentRequest):

At the agent executor, the decorator wraps invoke, and the active span is enriched with the agent's metadata:

@do_trace
def invoke(self, messages, config=None, thread_id="default", ...):
    active_span = trace.get_current_span()
    if active_span and hasattr(self.config, "metadata"):
        for key, value in self.config.metadata.model_dump().items():
            active_span.set_attribute(f"agent.metadata.{key}", str(value))
        active_span.set_attribute("agent.id", self.config.metadata.id)
    ...

In Cloud Trace the result is a nested tree running from the API call down through the executor and into the model and tool calls.

Shipping OpenTelemetry with the agent

Installing to Agent Engine creates a fresh Python environment, so the utility library and its OTel dependencies have to travel with the agent definition. The deployment service copies the library into a temporary directory and lists it as extra_packages:

def _setup_deployment_directory(self, agent_definition: str):
    temp_dir = tempfile.mkdtemp(prefix="agent_install_")
    temp_path = Path(temp_dir)
    yaml_path = temp_path / "agent_definition.yaml"
    yaml_path.write_text(agent_definition)
    import agent_utility
    lib_src = Path(agent_utility.__file__).parent
    shutil.copytree(lib_src, temp_path / "agent_utility")
    return temp_dir, "agent_definition.yaml", lib_src

There's a catch. Agent Engine doesn't install dependencies from local directory copies, so each one must also appear as an explicit package requirement:

runtime_deps = self._get_library_dependencies(lib_src)
requirements = [
"google-cloud-aiplatform[agent_engines]>=1.120.0",
"langchain>=0.2.0",
"langchain-google-vertexai>=2.0.0",
] + runtime_deps

The agent is then created with enable_tracing=True, which switches on the OpenInference LangGraph instrumentation:

agent = LanggraphAgent(
    model="gemini-2.0-flash",
    runnable_builder=runnable_builder_fn,
    enable_tracing=True,
)

Inside the runnable builder, the executor loads with tracing enabled. use_sync_nodes=True is required too, because Agent Engine's execution model is synchronous:

def runnable_builder_fn(model, *, tools=None, checkpointer=None, **kwargs):
    executor = load_agent_from_yaml(
        "agent.yaml",
        enable_tracing=True,
        use_sync_nodes=True,
    )
    return RunnableLambda(input_adapter) | executor.graph

Lessons from the setup

1. Agent Engine ships with no tracing for LangGraph. Everything has to be added.
2. Call the OpenTelemetry setup function at agent init. That's the entry point here, since no web server lifecycle exists.
3. Send OTLP/gRPC to telemetry.googleapis.com:443 and skip the collector.
4. Make the setup non-fatal. A missing IAM role shouldn't be able to take the agent down.
5. Redact credentials before export, because LangGraph state serialized into spans can contain secrets.
6. List OTel dependencies explicitly in requirements.
7. Pair the decorator with named spans so traces are searchable trees and not a pile of generic function names.

IAM roles for the service account

Full observability needs three roles on the Agent Engine service account:

• roles/cloudtrace.agent
• roles/logging.logWriter
• roles/monitoring.metricWriter

The setup code checks for Cloud Monitoring permissions before enabling the metric exporter. A missing monitoring.metricWriter therefore turns off metrics only, and traces keep flowing.