A2A × ADKの"観測粒度"を設計する - Langfuse & Cloud Trace でトレース構造を可視化 -

google-adk[a2a]
google-genai
langfuse
python-dotenv
opentelemetry-instrumentation-google-genai
opentelemetry-exporter-gcp-logging
opentelemetry-exporter-gcp-monitoring
opentelemetry-exporter-otlp-proto-grpc
opentelemetry-instrumentation-vertexai>=2.0b0

- Python: 3.12.4
- google-genai: 1.39.1
- langfuse: 3.5.2
- opentelemetry-instrumentation-google-genai: 0.1.5
- opentelemetry-instrumentation-vertexai: 0.1.11
- python-dotenv: 1.1.1

from datetime import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """指定された都市の天気情報を取得"""
    if city.lower() == "東京":
        return {
            "status": "success",
            "report": (
                "東京の天気は晴れで、気温は25度です。"
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"'{city}'の天気情報は利用できません。",
        }

def get_current_time(city: str) -> dict:
    """指定された都市の現在時刻を取得"""
    if city.lower() == "東京":
        tz_identifier = "Asia/Tokyo"
    else:
        return {
            "status": "error",
            "error_message": (
                f"'{city}'のタイムゾーン情報は利用できません。"
            ),
        }
    tz = ZoneInfo(tz_identifier)
    now = datetime.now(tz)
    report = (
        f'{city}の現在時刻は{now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}です。'
    )
    return {"status": "success", "report": report}

root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.5-flash-lite",
    description=(
        "東京の天気と現在時刻を取得するエージェントです。"
    ),
    instruction=(
        "東京の天気と現在時刻を取得するエージェントです。"
    ),
    tools=[get_weather, get_current_time],
)

import uvicorn
from google.adk.agents import ADKAgentExecutor
from a2a.server import A2AStarletteApplication, DefaultRequestHandler
from a2a.storage import InMemoryTaskStore

request_handler = DefaultRequestHandler(
    agent_executor=ADKAgentExecutor(agent=root_agent),
    task_store=InMemoryTaskStore(),
)

server = A2AStarletteApplication(
    agent_card=public_agent_card,
    http_handler=request_handler,
)

uvicorn.run(server.build(), host='0.0.0.0', port=9999)

from google.adk.a2a import to_a2a

# ADKエージェントをA2Aサーバーとして公開
a2a_app = to_a2a(root_agent, port=8001)

from google.adk.a2a import RemoteA2aAgent, AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents import Agent

weather_time_agent = RemoteA2aAgent(
    name="weather_time_agent",
    description=(
        "東京の天気と現在時刻を取得するエージェントです。"
    ),
    agent_card=f"http://localhost:8001/a2a/weather_time_agent{AGENT_CARD_WELL_KNOWN_PATH}",
)

root_agent = Agent(
    model="gemini-2.5-flash-lite",
    name="root_agent",
    instruction=(
        "日本語で回答して"
    ),
    sub_agents=[weather_time_agent],
)

{
  "capabilities": {},
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "description": "東京の天気と現在時刻を取得するエージェントです。",
  "name": "weather_time_agent",
  "skills": [
    {
      "id": "weather_time_checking",
      "name": "Weather and Time Checking",
      "description": "東京の天気と現在時刻を取得します。",
      "tags": ["weather", "time"]
    }
  ],
  "url": "http://localhost:8001/a2a/weather_time_agent",
  "version": "1.0.0"
}

from langfuse import get_client

langfuse = get_client()

import google.auth
import google.auth.transport.requests
import grpc
from opentelemetry import trace, logs, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.cloud_logging import CloudLoggingExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.cloud_monitoring import CloudMonitoringMetricsExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
from opentelemetry.instrumentation.vertexai import VertexAIInstrumentor
from opentelemetry.instrumentation.google_genai import GoogleGenAiSdkInstrumentor
from opentelemetry.exporter.otlp.proto.grpc._auth import AuthMetadataPlugin

SERVICE_NAME = ResourceAttributes.SERVICE_NAME

# OpenTelemetry初期化
def initialize_opentelemetry() -> None:
    """OpenTelemetryの初期化とGoogle Cloud Observabilityの設定"""
    credentials, project_id = google.auth.default()
    resource = Resource.create(
        attributes={
            SERVICE_NAME: "multi-tool-agent",
            # The project to send spans to
            "gcp.project_id": project_id,
        }
    )
    # Set up OTLP auth
    request = google.auth.transport.requests.Request()
    auth_metadata_plugin = AuthMetadataPlugin(credentials=credentials, request=request)
    channel_creds = grpc.composite_channel_credentials(
        grpc.ssl_channel_credentials(),
        grpc.metadata_call_credentials(auth_metadata_plugin),
    )
    # Set up OpenTelemetry Python SDK
    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(
                credentials=channel_creds,
                endpoint="https://telemetry.googleapis.com:443/v1/traces",
            )
        )
    )

    trace.set_tracer_provider(tracer_provider)
    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(CloudLoggingExporter())
    )
    logs.set_logger_provider(logger_provider)
    reader = PeriodicExportingMetricReader(CloudMonitoringMetricsExporter())
    meter_provider = MeterProvider(metric_readers=[reader], resource=resource)
    metrics.set_meter_provider(meter_provider)
    # Load instrumentors
    # ADK uses Vertex AI and Google Gen AI SDKs.
    VertexAIInstrumentor().instrument()
    GoogleGenAiSdkInstrumentor().instrument()

    print(f"OpenTelemetryが初期化されました。Google Cloud Project ID: {project_id}")

# OpenTelemetryの初期化を実行
initialize_opentelemetry()

from langfuse import Langfuse

# フィルタリング機能使用時
langfuse = Langfuse(
    blocked_instrumentation_scopes=["a2a-python-sdk", "gcp.vertex.agent"]
)

langfuse = Langfuse(blocked_instrumentation_scopes=["a2a-python-sdk"])