Publish to Gemini and configure agent observability

Register a deployed agent with Gemini Enterprise and understand the separate paths for trace metadata and completion payloads. The local lab validates registration settings, endpoint selection and observability configuration; operational checks confirm delivery after a real deployment.

Problem

When you need this. After agents-cli deploy rolls out your agent to Agent Runtime or Cloud Run, internal users still need a way to invoke it from Gemini Enterprise, and operators need a way to inspect production prompts, tool calls, and latency without leaking sensitive prompt payloads into Cloud Trace span attributes. In this final lesson of the course, you will register docuctl-docs with Gemini using agents-cli publish gemini (comparing ADK :streamQuery registration against A2A .well-known/agent-card.json registration) and configure the two-tier OpenTelemetry, Cloud Storage, and BigQuery observability pipeline generated in telemetry.tf.

Concepts

Publishing and observing a production agent involves two distinct boundaries: registration with Discovery Engine (publish gemini) and two-tier telemetry (telemetry.tf).

ADK vs A2A registration (--registration-type)

adk mode registers an Agent Runtime ReasoningEngine invoked via :streamQuery; a2a mode fetches an A2A .well-known/agent-card.json endpoint (on Cloud Run, GKE, or Agent Runtime) and registers its JSON-RPC skills.

Host-aware token selection

When fetching --agent-card-url, publish gemini sends an OAuth access token to *.googleapis.com URLs and an audience-scoped OIDC ID token to Cloud Run or custom HTTPS URLs so external endpoints never receive a broad GCP access token.

Tier 1 telemetry (NO_CONTENT spans)

OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=NO_CONTENT and ADK_CAPTURE_MESSAGE_CONTENT_IN_SPANS=false keep message content out of the standard instrumentation’s span attributes while retaining timing and usage metadata. ADK 2.13 defaults its own capture switch to true when unset. Review custom attributes separately.

Tier 2 telemetry (GCS JSONL + completions_view)

OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK=upload writes full prompt/response messages as .jsonl files to gs://<logs_bucket>/completions and logs a gen_ai.client.inference.operation.details pointer that BigQuery joins in completions_view.

Architecture

Registration connects a deployed endpoint to Gemini Enterprise. Observability follows a separate path: trace metadata goes to Cloud Trace, while the configured completion hook sends message payloads to Cloud Storage for subsequent analysis. The local helper checks routing and configuration; it does not register an agent or verify telemetry delivery.

Working code

Download the complete lesson lab, extract it, and open its agents-cli-publish-observe directory. It includes the source, tests and fixtures. With Python 3.12+ and uv installed, prepare the local environment:

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt
LAB_DIR="$PWD"

Replace the project path below with the directory created in Lesson 1. Use the model credentials configured there for model activities; use your cloud credentials for cloud activities.

export PROJECT_DIR="/absolute/path/to/docuctl-docs"
cd "$PROJECT_DIR"
export AGENT_CARD_URL="YOUR_DEPLOYED_AGENT_CARD_URL"
export GEMINI_APP_ID="YOUR_GEMINI_ENTERPRISE_APP_RESOURCE_NAME"

Continue from the deployed project in Provision Terraform and deploy with Agents CLI. Registration needs a real endpoint, a Gemini Enterprise app and the required IAM permissions. Set AGENT_CARD_URL to the card URL from your deployment and GEMINI_APP_ID to the app resource name returned by your project. The supplied Python helper checks configuration locally; it does not register an agent or deliver telemetry.

Register your agent with publish gemini

To discover available Gemini Enterprise intranet apps (APP_TYPE_INTRANET) in your project before registering an agent, run agents-cli publish gemini --list:

# List Gemini Enterprise apps in the project
agents-cli publish gemini --list --project-id "$GOOGLE_CLOUD_PROJECT"

# Register a Cloud Run A2A agent explicitly in programmatic mode
agents-cli publish gemini \
  --registration-type a2a \
  --agent-card-url $AGENT_CARD_URL \
  --gemini-app-id $GEMINI_APP_ID \
  --display-name "docuctl Docs Assistant" \
  --description "Answers developer questions about docuctl 2.x commands"

Note that agents-cli publish gemini sends all human-readable progress tables to stderr so stdout carries only machine-readable JSON for CI/CD scripts. Based on the location segment in --gemini-app-id (projects/<number>/locations/<location>/collections/<collection>/engines/<engine_id>), the CLI routes the registration request to https://discoveryengine.googleapis.com (when location is global) or https://<location>-discoveryengine.googleapis.com (such as https://us-discoveryengine.googleapis.com).

Choose ADK or A2A registration mode

agents-cli publish gemini supports two registration modes via --registration-type:

When publish gemini fetches the A2A Agent Card from --agent-card-url to register its skills with Discovery Engine, it chooses the authentication token type based on the target hostname:

• *.googleapis.com URLs (such as the Agent Runtime A2A passthrough endpoint): sends a Google Cloud OAuth access token (Authorization: Bearer <access_token>).
• Cloud Run, GKE, or custom HTTPS domains: sends an audience-scoped OIDC ID token bound to the target URL so an external endpoint never receives a broad GCP access token.

Here is publish_observe.py, a local model of the Discovery Engine endpoint router, the Agent Card token-type selector, the programmatic registration resolver, and the two-tier observability validator:

"""Gemini registration resolver and two-tier observability validator for agents-cli."""

from __future__ import annotations

from typing import Any
from urllib.parse import urlparse


def get_discovery_engine_endpoint(location: str) -> str:
    """Return the regional or global Discovery Engine endpoint (mirrors cmd_publish.py)."""
    if location == "global":
        return "https://discoveryengine.googleapis.com"
    return f"https://{location}-discoveryengine.googleapis.com"


def select_card_auth_token_type(agent_card_url: str) -> str:
    """Select access_token for *.googleapis.com and audience_id_token for Cloud Run / external URLs."""
    parsed = urlparse(agent_card_url)
    hostname = parsed.hostname or ""
    if hostname.endswith(".googleapis.com"):
        return "access_token"
    return "audience_id_token"


def resolve_publish_gemini_request(
    *,
    registration_type: str | None,
    gemini_app_id: str | None,
    agent_runtime_id: str | None = None,
    agent_card_url: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> dict[str, str]:
    """Validate programmatic `agents-cli publish gemini` arguments and build registration config."""
    if not registration_type and metadata:
        if metadata.get("is_a2a"):
            registration_type = "a2a"
        elif metadata.get("remote_agent_runtime_id"):
            registration_type = "adk"

    if not registration_type:
        raise ValueError(
            "--registration-type is required in programmatic mode. Rerun with one of:\n"
            "  --registration-type <a2a|adk>  a2a needs --agent-card-url, adk needs --agent-runtime-id\n"
            "  -i, --interactive              be prompted for each missing value"
        )

    if not gemini_app_id:
        raise ValueError("--gemini-app-id is required in programmatic mode.")

    parts = gemini_app_id.split("/")
    if len(parts) != 8 or parts[0] != "projects" or parts[2] != "locations" or parts[6] != "engines":
        raise ValueError(
            f"Invalid --gemini-app-id {gemini_app_id!r}; expected "
            "projects/<number>/locations/<location>/collections/<collection>/engines/<engine_id>."
        )
    location = parts[3]
    endpoint = get_discovery_engine_endpoint(location)

    if registration_type == "adk":
        runtime_id = agent_runtime_id or (metadata or {}).get("remote_agent_runtime_id")
        if not runtime_id:
            raise ValueError("ADK registration requires --agent-runtime-id or deployment_metadata.json.")
        return {
            "registration_type": "adk",
            "discovery_engine_endpoint": endpoint,
            "target_resource": runtime_id,
            "protocol": "streamQuery",
        }

    if registration_type == "a2a":
        card_url = agent_card_url
        if not card_url and metadata and metadata.get("remote_agent_runtime_id"):
            rt = metadata["remote_agent_runtime_id"]
            rt_parts = rt.split("/")
            rt_region = rt_parts[3]
            card_url = f"https://{rt_region}-aiplatform.googleapis.com/v1beta1/{rt}/a2a/v1/card"
        if not card_url:
            raise ValueError("A2A registration requires --agent-card-url or A2A deployment_metadata.json.")
        return {
            "registration_type": "a2a",
            "discovery_engine_endpoint": endpoint,
            "agent_card_url": card_url,
            "card_auth_token_type": select_card_auth_token_type(card_url),
            "protocol": "JSONRPC",
        }

    raise ValueError(f"Unsupported registration_type: {registration_type!r}")


def validate_observability_env(env: dict[str, str]) -> dict[str, Any]:
    """Validate the two-tier OpenTelemetry + GCS/BigQuery completions configuration."""
    span_capture = env.get("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "NO_CONTENT")
    adk_span_capture = env.get("ADK_CAPTURE_MESSAGE_CONTENT_IN_SPANS", "true").lower() == "true"
    hook = env.get("OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK", "")
    bucket = env.get("LOGS_BUCKET_NAME", "").strip()
    upload_path = env.get("OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH", "").strip()

    if hook == "upload" and (not bucket or not upload_path.startswith("gs://")):
        raise ValueError(
            "OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK='upload' requires LOGS_BUCKET_NAME "
            "and a gs:// OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH; otherwise prompt/response "
            "JSONL payloads are dropped while spans record NO_CONTENT."
        )

    return {
        "spans_include_message_content": (span_capture != "NO_CONTENT") or adk_span_capture,
        "gcs_completion_upload_enabled": hook == "upload" and bool(bucket),
        "upload_base_path": upload_path,
        "bigquery_join_table": "completions_view",
    }

Configure two-tier agent observability

Why do scaffolded agents-cli projects configure both OpenTelemetry span flags and GCS/BigQuery resources in telemetry.tf? Because production agent telemetry separates latency tracing from full prompt/response inspection:

1. Tier 1 — Cloud Trace spans (NO_CONTENT): In service.tf, OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT is set to "NO_CONTENT" and ADK_CAPTURE_MESSAGE_CONTENT_IN_SPANS is set to "false". Cloud Trace spans record invocation timing, model names, token counts, and tool names without storing multi-kilobyte prompt or documentation payloads inside span attributes (which have size limits and their own access controls).
2. Tier 2 — GCS JSONL uploads + Cloud Logging sink + BigQuery completions_view: Setting OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK=upload and OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH=gs://<logs_bucket>/completions instructs the OpenTelemetry GenAI instrumentation to upload full prompt, tool-call, and response messages as newline-delimited JSON (.jsonl) to Cloud Storage and emit a structured Cloud Logging entry (labels."event.name"="gen_ai.client.inference.operation.details") containing the GCS file references (gen_ai.input.messages_ref and gen_ai.output.messages_ref).

Inspect the excerpt from the scaffolded telemetry.tf that wires the Cloud Logging sink, the GCS external table, and the BigQuery table together:

# Log sink to route GenAI telemetry logs directly to BigQuery
resource "google_logging_project_sink" "genai_logs_to_bq" {
  name        = "${var.project_name}-genai-logs"
  project     = var.project_id
  destination = "bigquery.googleapis.com/projects/${var.project_id}/datasets/${google_bigquery_dataset.telemetry_dataset.dataset_id}"
  filter      = "labels.\"event.name\"=\"gen_ai.client.inference.operation.details\" AND (labels.\"gen_ai.input.messages_ref\" =~ \".*${var.project_name}.*\" OR labels.\"gen_ai.output.messages_ref\" =~ \".*${var.project_name}.*\")"

  unique_writer_identity = true

  bigquery_options {
    use_partitioned_tables = true
  }
}

# External table for completions data (messages/parts) stored in GCS
resource "google_bigquery_table" "completions_external_table" {
  project    = var.project_id
  dataset_id = google_bigquery_dataset.telemetry_dataset.dataset_id
  table_id   = "completions"

  external_data_configuration {
    autodetect    = false
    source_format = "NEWLINE_DELIMITED_JSON"
    source_uris   = ["gs://${google_storage_bucket.logs_data_bucket.name}/completions/*"]
    connection_id = google_bigquery_connection.genai_telemetry_connection.name
  }
}

# Pre-create the log export table so completions_view works on first deploy
resource "google_bigquery_table" "genai_logs_table" {
  project    = var.project_id
  dataset_id = google_bigquery_dataset.telemetry_dataset.dataset_id
  table_id   = "gen_ai_client_inference_operation_details"

  time_partitioning {
    type  = "DAY"
    field = "timestamp"
  }

  schema = file("${path.module}/../shared/genai_logs_schema.json")

  lifecycle {
    ignore_changes = [schema]
  }
}

Notice lifecycle { ignore_changes = [schema] } on google_bigquery_table.genai_logs_table (gen_ai_client_inference_operation_details): as Cloud Logging exports new log entries, BigQuery automatically evolves the table schema to add newly observed label columns. Ignoring schema changes in Terraform prevents a subsequent terraform apply from treating those Cloud Logging columns as drift and trying to reconcile them to the template schema. Review each plan; whether a schema change updates or replaces a table depends on the change and provider behavior.

Inspect the supplied configuration examples and local observability validation in recorded-publish-observe.json:

{
  "cli_version": "1.11.0",
  "project": "vertex-ai-experminent",
  "project_number": "913703049174",
  "publish_modes": {
    "a2a_cloud_run": {
      "command": "agents-cli publish gemini --registration-type a2a --agent-card-url https://shipctl-docs-913703049174.us-central1.run.app/a2a/app/.well-known/agent-card.json --gemini-app-id projects/913703049174/locations/global/collections/default_collection/engines/eng-ops-assistant --display-name \"shipctl Docs Assistant\"",
      "resolved": {
        "registration_type": "a2a",
        "discovery_engine_endpoint": "https://discoveryengine.googleapis.com",
        "agent_card_url": "https://shipctl-docs-913703049174.us-central1.run.app/a2a/app/.well-known/agent-card.json",
        "card_auth_token_type": "audience_id_token",
        "protocol": "JSONRPC"
      }
    },
    "a2a_agent_runtime_from_metadata": {
      "metadata_file": "deployment_metadata.json",
      "resolved": {
        "registration_type": "a2a",
        "discovery_engine_endpoint": "https://us-discoveryengine.googleapis.com",
        "agent_card_url": "https://us-central1-aiplatform.googleapis.com/v1beta1/projects/913703049174/locations/us-central1/reasoningEngines/7482910384756192000/a2a/v1/card",
        "card_auth_token_type": "access_token",
        "protocol": "JSONRPC"
      }
    },
    "adk_agent_runtime": {
      "command": "agents-cli publish gemini --registration-type adk --agent-runtime-id projects/913703049174/locations/us-central1/reasoningEngines/7482910384756192000 --gemini-app-id projects/913703049174/locations/global/collections/default_collection/engines/eng-ops-assistant",
      "resolved": {
        "registration_type": "adk",
        "discovery_engine_endpoint": "https://discoveryengine.googleapis.com",
        "target_resource": "projects/913703049174/locations/us-central1/reasoningEngines/7482910384756192000",
        "protocol": "streamQuery"
      }
    }
  },
  "observability_tiers": {
    "env": {
      "LOGS_BUCKET_NAME": "vertex-ai-experminent-shipctl-docs-logs",
      "OTEL_SERVICE_NAME": "shipctl-docs",
      "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "NO_CONTENT",
      "ADK_CAPTURE_MESSAGE_CONTENT_IN_SPANS": "false",
      "OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental",
      "OTEL_INSTRUMENTATION_GENAI_UPLOAD_FORMAT": "jsonl",
      "OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK": "upload",
      "OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH": "gs://vertex-ai-experminent-shipctl-docs-logs/completions"
    },
    "validated": {
      "spans_include_message_content": false,
      "gcs_completion_upload_enabled": true,
      "upload_base_path": "gs://vertex-ai-experminent-shipctl-docs-logs/completions",
      "bigquery_join_table": "completions_view"
    }
  },
  "failures": {
    "missing_registration_type": {
      "command": "agents-cli publish gemini --project-id vertex-ai-experminent",
      "exit_code": 2,
      "error": "--registration-type is required in programmatic mode. Rerun with one of:\n  --registration-type <a2a|adk>  a2a needs --agent-card-url, adk needs --agent-runtime-id\n  -i, --interactive              be prompted for each missing value"
    },
    "upload_hook_without_logs_bucket": {
      "env": {
        "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "NO_CONTENT",
        "OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK": "upload",
        "LOGS_BUCKET_NAME": "",
        "OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH": ""
      },
      "error": "OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK='upload' requires LOGS_BUCKET_NAME and a gs:// OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH; otherwise prompt/response JSONL payloads are dropped while spans record NO_CONTENT."
    }
  }
}

After deployment, make a test request with non-sensitive input. Confirm that the registered agent is discoverable, the request reaches the intended endpoint, the trace appears, and payload references resolve for an authorized operator. Check retention and access policies before capturing real user content. Setting environment variables or passing the local validator proves none of those delivery outcomes.

Failure cases

Run the offline verification suite to test ADK and A2A registration resolution, token selection, and the two failure boundaries:

cd "$LAB_DIR"
"$LAB_DIR/.venv/bin/python" -m unittest -v test_publish_observe.py

Watch for two failure modes when automating registration and telemetry:

Programmatic publish gemini without –registration-type. When agents-cli publish gemini runs in CI/CD without -i (--interactive) and without a local deployment_metadata.json file, omitting --registration-type exits immediately with code 2:

Error: --registration-type is required in programmatic mode. Rerun with one of:
  --registration-type <a2a|adk>  a2a needs --agent-card-url, adk needs --agent-runtime-id
  -i, --interactive              be prompted for each missing value

Always pass --registration-type a2a --agent-card-url ... (or --registration-type adk --agent-runtime-id ...) explicitly in non-interactive pipelines.

Setting OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK=upload without LOGS_BUCKET_NAME. Because Tier 1 (OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=NO_CONTENT) and Tier 2 (OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK=upload) are configured through separate environment variables, enabling the upload hook without providing LOGS_BUCKET_NAME and a gs:// OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH leaves the hook without a destination bucket while Cloud Trace still omits message content. Validate that LOGS_BUCKET_NAME and OTEL_INSTRUMENTATION_GENAI_UPLOAD_BASE_PATH are populated whenever OTEL_INSTRUMENTATION_GENAI_COMPLETION_HOOK=upload is enabled.

Deployment

After deployment, register the intended endpoint and retain the registration result alongside deployment metadata. Send a non-sensitive smoke request, confirm that the registered agent is discoverable, and verify the trace and payload references with an authorized operator. Configuration validation alone cannot establish that these services received the data.

Scaling

Full message payloads can grow much faster than trace metadata. Set access and retention policies for both storage paths, measure ingestion and query volume, and test how missing payload references appear to operators. Choose query filters using the actual table schema and partitioning; this lab does not measure production ingestion or query performance.

Key takeaways

• agents-cli publish gemini supports adk mode (:streamQuery on Agent Runtime) and a2a mode (Agent Card JSON-RPC across Cloud Run, GKE, or Agent Runtime).
• Fetching an A2A Agent Card uses an OAuth access token for *.googleapis.com hosts and an audience-scoped OIDC ID token for Cloud Run or custom domains.
• Two-tier observability pairs metadata-only Cloud Trace spans (NO_CONTENT) with GCS .jsonl completion uploads joined in BigQuery completions_view.

Best practices

• Keep lifecycle { ignore_changes = [schema] } on google_bigquery_table.genai_logs_table in telemetry.tf and review the plan when Cloud Logging adds columns.
• Enforce IAM bucket-level access control on the GCS logs bucket (LOGS_BUCKET_NAME), since Tier 2 .jsonl files contain raw user prompts and tool outputs.
• Parse stdout (JSON) separately from stderr (status tables) when running agents-cli publish gemini in automated pipelines.

References

Commands reviewed against Google Agents CLI 1.11.0 and the linked official documentation on October 23, 2026.

• Google Agents CLI publish and observability guides: agents-cli publish gemini, ADK vs A2A registration, and BigQuery Agent Analytics (--bq-analytics).
• OpenTelemetry GenAI semantic conventions: gen_ai.client.inference.operation.details events and completion upload hooks.