logging-callback-setup

logging-callback-setup

Hook structured logging into the ADK 2.0 callback chain so every model call, tool call, and state change is captured.

Callback hooks

Hook	Fires
on_session_created	New session starts
on_before_model_call	Before each LLM request
on_after_model_call	After LLM response
on_before_tool_call	Before tool fires
on_after_tool_call	After tool returns
on_event_emitted	Any event added to session

Structured stdout (dev)

import json, time
from google.adk.callbacks import (
    on_before_model_call,
    on_after_model_call,
    on_before_tool_call,
    on_after_tool_call,
)

def log(level, **fields):
    print(json.dumps({"ts": time.time(), "level": level, **fields}))

@on_before_model_call
async def log_model_in(ctx, request):
    log("INFO", event="model_call_start", session=ctx.session.id, model=request.model)
    return request

@on_after_model_call
async def log_model_out(ctx, response):
    log("INFO", event="model_call_end", session=ctx.session.id,
        tokens_in=response.usage.input_tokens, tokens_out=response.usage.output_tokens)
    return response

@on_before_tool_call
async def log_tool_in(ctx, tool_name, args):
    log("INFO", event="tool_call_start", session=ctx.session.id, tool=tool_name, args=args)

@on_after_tool_call
async def log_tool_out(ctx, tool_name, args, result):
    log("INFO", event="tool_call_end", session=ctx.session.id, tool=tool_name,
        ok="error" not in (result or {}))

Cloud Logging (production)

from google.cloud import logging as cloud_logging
client = cloud_logging.Client()
logger = client.logger("adk-agent")

@on_after_tool_call
async def cloud_log(ctx, tool_name, args, result):
    logger.log_struct({
        "event": "tool_call",
        "session_id": ctx.session.id,
        "user_id": ctx.session.user_id,
        "tool": tool_name,
        "args": args,
        "ok": "error" not in (result or {}),
    }, severity="INFO")

OpenTelemetry tracing

from opentelemetry import trace
tracer = trace.get_tracer("adk-agent")

@on_before_model_call
async def trace_model(ctx, request):
    span = tracer.start_span("model_call",
        attributes={"model": request.model, "session.id": ctx.session.id})
    ctx.runtime["model_span"] = span
    return request

@on_after_model_call
async def end_trace_model(ctx, response):
    span = ctx.runtime.pop("model_span", None)
    if span:
        span.set_attribute("tokens.in", response.usage.input_tokens)
        span.set_attribute("tokens.out", response.usage.output_tokens)
        span.end()
    return response

PII redaction

import re
def redact(s: str) -> str:
    s = re.sub(r"[\w.-]+@[\w.-]+", "[email]", s)
    s = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[ssn]", s)
    return s

@on_before_model_call
async def redact_logs(ctx, request):
    log("INFO", event="model_call_start", prompt=redact(request.prompt[:200]))
    return request

Validation

• Logs serialize cleanly (no exception traces in JSON output)
• Trace IDs propagate across callbacks (use OTel context)
• PII redaction tested with synthetic samples
• Volume manageable in production (consider sampling for high-traffic)

See also

• tool-monitor-dashboard for visualizing these logs
• agent-runtime-config for callback registration