feat(adk): expose gen_ai.client.token.usage Prometheus metric

docs/genai-token-metrics.md

@@ -0,0 +1,88 @@
+# GenAI token-usage metrics
+
+kagent's **Go ADK** agent runtime records the OpenTelemetry GenAI-semconv metric
+[`gen_ai.client.token.usage`](https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiclienttokenusage)
+using the native Prometheus client library and exposes it for scraping. It lets you graph and alert
+on token spend per model / provider without parsing traces.
+
+## What is emitted
+
+A Prometheus histogram, served at **`/metrics`** on the agent's HTTP port:
+
+| Prometheus name | OTel semconv | Notes |
+| --- | --- | --- |
+| `gen_ai_client_token_usage` | `gen_ai.client.token.usage` | histogram, semconv-recommended buckets |
+
+Labels (semconv attributes, dots → underscores):
+
+| Label | Values |
+| --- | --- |
+| `gen_ai_token_type` | `input`, `output` (output = candidate + reasoning tokens) |
+| `gen_ai_operation_name` | `chat` |
+| `gen_ai_provider_name` | well-known value, e.g. `openai`, `anthropic`, `gcp.vertex_ai`, `aws.bedrock`, `azure.ai.openai` |
+| `gen_ai_request_model` | configured model, e.g. `gpt-4o` |
+| `gen_ai_response_model` | model the provider served (falls back to request model) |
+| `error_type` | set on failed requests; empty otherwise |
+
+One observation is recorded per LLM call (streaming partial chunks are not double-counted).
+
+## Configuration
+
+- **Runtime**: available on Declarative agents with `runtime: go`. (The Python runtime is not yet
+  instrumented.)
+- **No flag to enable**: the `/metrics` endpoint is always served, and the controller adds Prometheus
+  pod annotations to Go-runtime agent Deployments automatically:
+
+  ```yaml
+  metadata:
+    annotations:
+      prometheus.io/scrape: "true"
+      prometheus.io/port: "<agent-port>"
+      prometheus.io/path: "/metrics"
+  ```
+
+### Scraping it
+
+Any Prometheus-compatible scraper that honors pod annotations will pick agents up. With an
+OpenTelemetry Collector, add a `prometheus` receiver job with pod discovery:
+
+```yaml
+receivers:
+  prometheus:
+    config:
+      scrape_configs:
+        - job_name: kagent-agents
+          kubernetes_sd_configs: [{ role: pod }]
+          relabel_configs:
+            - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
+              regex: "true"
+              action: keep
+            - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
+              target_label: __metrics_path__
+            - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port]
+              regex: ([^:]+)(?::\d+)?;(\d+)
+              replacement: $$1:$$2
+              target_label: __address__
+```
+
+## Verifying
+
+```bash
+# 1. exec into a Go-runtime agent pod and curl its metrics endpoint
+kubectl exec <go-agent-pod> -- wget -qO- localhost:<agent-port>/metrics | grep gen_ai_client_token_usage
+
+# 2. after chatting with the agent you should see series like:
+# gen_ai_client_token_usage_count{gen_ai_token_type="input",gen_ai_provider_name="gcp.vertex_ai",
+#   gen_ai_request_model="gemini-2.5-flash",gen_ai_response_model="gemini-2.5-flash",
+#   gen_ai_operation_name="chat",error_type=""} 1
+```
+
+<!-- TODO: add a Grafana/Prometheus screenshot of gen_ai_client_token_usage once a cluster with a
+     scrape target is available. -->
+
+## Follow-ups
+
+- Optional OTLP **push** (in addition to the scrape endpoint) via the OpenTelemetry
+  [Prometheus→OTLP bridge](https://pkg.go.dev/go.opentelemetry.io/contrib/bridges/prometheus), for
+  environments that push to an OTLP collector rather than scrape.
+- Python ADK runtime instrumentation.

go/adk/cmd/main.go

@@ -19,6 +19,7 @@ import (
 	runnerpkg "github.com/kagent-dev/kagent/go/adk/pkg/runner"
 	"github.com/kagent-dev/kagent/go/adk/pkg/session"
 	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
+	"github.com/kagent-dev/kagent/go/api/adk"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -185,13 +186,16 @@ func main() {
 	}
 
 	stream := agentConfig.GetStream()
+	modelName, providerName := resolveModelLabels(agentConfig)
 	executor := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{
 		RunnerConfig:       runnerConfig,
 		SubagentSessionIDs: subagentSessionIDs,
 		SessionService:     sessionService,
 		Stream:             stream,
 		AppName:            appName,
 		Logger:             logger,
+		ModelName:          modelName,
+		ProviderName:       providerName,
 	})
 
 	// Build the agent card.
@@ -231,6 +235,16 @@ func main() {
 	}
 }
 
+// resolveModelLabels derives the gen_ai.request.model / gen_ai.provider.name
+// labels for token-usage metrics from the agent config. Returns empty strings
+// when no model is configured; the metric simply omits those attributes.
+func resolveModelLabels(agentConfig *adk.AgentConfig) (model, provider string) {
+	if agentConfig == nil || agentConfig.Model == nil {
+		return "", ""
+	}
+	return config.ModelName(agentConfig.Model), telemetry.SemconvProviderName(agentConfig.Model.GetType())
+}
+
 func deriveAppName(kagentName, kagentNamespace string, agentCard *a2atype.AgentCard, logger logr.Logger) string {
 	if kagentNamespace != "" && kagentName != "" {
 		namespace := strings.ReplaceAll(kagentNamespace, "-", "_")

go/adk/pkg/a2a/executor.go

@@ -20,6 +20,7 @@ import (
 	adkagent "google.golang.org/adk/agent"
 	"google.golang.org/adk/runner"
 	"google.golang.org/adk/server/adka2a" //nolint:staticcheck // kagent still uses a2a-go v1; this ADK package is the compatibility adapter.
+	adksession "google.golang.org/adk/session"
 )
 
 const (
@@ -37,6 +38,11 @@ type KAgentExecutorConfig struct {
 	AppName            string
 	SkillsDirectory    string
 	Logger             logr.Logger
+	// ModelName and ProviderName label GenAI token-usage metrics
+	// (gen_ai.request.model / gen_ai.provider.name). Both may be empty, in
+	// which case the corresponding metric attributes are omitted.
+	ModelName    string
+	ProviderName string
 }
 
 // KAgentExecutor implements a2asrv.AgentExecutor
@@ -48,6 +54,8 @@ type KAgentExecutor struct {
 	appName            string
 	skillsDirectory    string
 	logger             logr.Logger
+	modelName          string
+	providerName       string
 }
 
 var _ a2asrv.AgentExecutor = (*KAgentExecutor)(nil)
@@ -69,9 +77,31 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 		appName:            cfg.AppName,
 		skillsDirectory:    skillsDir,
 		logger:             cfg.Logger.WithName("kagent-executor"),
+		modelName:          cfg.ModelName,
+		providerName:       cfg.ProviderName,
 	}
 }
 
+// recordTokenUsage records GenAI token usage for a single agent event on the
+// gen_ai.client.token.usage histogram. Partial (streaming) events are skipped:
+// a streamed LLM call emits many Partial chunks but usage is reported once on
+// the aggregated non-partial event, so this counts one observation per LLM
+// call, not per stream chunk. Output combines candidate + reasoning tokens.
+func (e *KAgentExecutor) recordTokenUsage(adkEvent *adksession.Event) {
+	um := adkEvent.UsageMetadata
+	if um == nil || adkEvent.Partial {
+		return
+	}
+	telemetry.RecordTokenUsage(telemetry.TokenUsage{
+		RequestModel:  e.modelName,
+		ResponseModel: adkEvent.ModelVersion,
+		Provider:      e.providerName,
+		ErrorType:     adkEvent.ErrorCode,
+		InputTokens:   int64(um.PromptTokenCount),
+		OutputTokens:  int64(um.CandidatesTokenCount) + int64(um.ThoughtsTokenCount),
+	})
+}
+
 // UserIDCallInterceptor returns an a2asrv.CallInterceptor that extracts the
 // x-user-id HTTP header from the incoming request metadata and sets it as the
 // authenticated user on the CallContext.
@@ -288,6 +318,9 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont
 		// Build per-event metadata (inherits baseMeta + adds invocation_id, usage etc.).
 		eventMeta := buildEventMeta(baseMeta, adkEvent)
 
+		// Record GenAI token usage for this event.
+		e.recordTokenUsage(adkEvent)
+
 		// Convert GenAI parts → A2A parts (with kagent stamping).
 		if adkEvent.Content == nil || len(adkEvent.Content.Parts) == 0 {
 			// Events with no content carry metadata only; still track invocationID/usage.

go/adk/pkg/a2a/executor_metrics_test.go

@@ -0,0 +1,56 @@
+package a2a
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	adkmodel "google.golang.org/adk/model"
+	adksession "google.golang.org/adk/session"
+	"google.golang.org/genai"
+)
+
+// gatherTokenUsageCount returns the total number of observations recorded on the
+// gen_ai_client_token_usage histogram (summed across all label series).
+func gatherTokenUsageCount(t *testing.T) uint64 {
+	t.Helper()
+	mfs, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatalf("gather: %v", err)
+	}
+	var total uint64
+	for _, mf := range mfs {
+		if mf.GetName() != "gen_ai_client_token_usage" {
+			continue
+		}
+		for _, m := range mf.GetMetric() {
+			total += m.GetHistogram().GetSampleCount()
+		}
+	}
+	return total
+}
+
+// TestExecutor_StreamingUsageNotDoubleCounted verifies that streamed LLM calls
+// record one input + one output observation per call, not one per stream chunk.
+// Partial events are skipped even when they carry usage metadata.
+func TestExecutor_StreamingUsageNotDoubleCounted(t *testing.T) {
+	e := &KAgentExecutor{modelName: "gemini-2.5-flash", providerName: "gcp.vertex_ai"}
+	usage := &genai.GenerateContentResponseUsageMetadata{PromptTokenCount: 10, CandidatesTokenCount: 5}
+
+	before := gatherTokenUsageCount(t)
+
+	const numCalls, chunksPerCall = 3, 4
+	for range numCalls {
+		// Streaming partial chunks that also carry usage — must be skipped.
+		for range chunksPerCall {
+			e.recordTokenUsage(&adksession.Event{LLMResponse: adkmodel.LLMResponse{Partial: true, UsageMetadata: usage}})
+		}
+		// Final aggregated (non-partial) event — the one that counts.
+		e.recordTokenUsage(&adksession.Event{LLMResponse: adkmodel.LLMResponse{UsageMetadata: usage}})
+	}
+
+	// One input + one output observation per LLM call, regardless of chunk count.
+	if got, want := gatherTokenUsageCount(t)-before, uint64(numCalls*2); got != want {
+		t.Fatalf("token usage observations = %d, want %d (must count %d LLM calls, not %d stream chunks)",
+			got, want, numCalls, numCalls*(chunksPerCall+1))
+	}
+}

go/adk/pkg/a2a/server/server.go

@@ -14,6 +14,8 @@ import (
 	"github.com/a2aproject/a2a-go/a2asrv"
 	"github.com/go-logr/logr"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+
+	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 )
 
 // ServerConfig holds configuration for the A2A server.
@@ -38,6 +40,8 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo
 
 	mux := http.NewServeMux()
 	RegisterHealthEndpoints(mux)
+	// Expose GenAI token-usage (and standard Go/process) metrics for scraping.
+	mux.Handle("/metrics", telemetry.MetricsHandler())
 	mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(&agentCard))
 	mux.Handle("/", jsonrpcHandler)
 	// Wrap the whole server mux to enable trace context extraction and an inbound
@@ -50,7 +54,7 @@ func NewA2AServer(agentCard a2atype.AgentCard, executor a2asrv.AgentExecutor, lo
 		}),
 		otelhttp.WithFilter(func(r *http.Request) bool {
 			switch r.URL.Path {
-			case "/health", "/healthz", a2asrv.WellKnownAgentCardPath:
+			case "/health", "/healthz", "/metrics", a2asrv.WellKnownAgentCardPath:
 				return false
 			default:
 				return true

go/adk/pkg/config/config_usage.go

@@ -129,6 +129,13 @@ func GetAgentConfigSummary(config *adk.AgentConfig) string {
 	return summary
 }
 
+// ModelName returns the configured model identifier (e.g. "gpt-4o",
+// "claude-3-5-sonnet"), or "unknown" for an unrecognized model type. It is the
+// exported form of getModelName, used to label GenAI telemetry.
+func ModelName(m adk.Model) string {
+	return getModelName(m)
+}
+
 func getModelName(m adk.Model) string {
 	switch m := m.(type) {
 	case *adk.OpenAI: