ADK 用 Google Cloud Eventarc ツール¶

ADKでサポートPython v2.6.0試験運用版 (Experimental)

EventarcToolset を使用すると、エージェントは Google Cloud Eventarc と相互作用し、構造化された CloudEvents を Eventarc メッセージバス (Message Bus) に非同期で発行できます。このツールセットは呼び出し間での組み込みコネクションプーリングとキャッシュを提供し、汎用的なイベント発行とドメイン固有のスキーマ適用イベントツールの両方をサポートします。

試験運用版 (Experimental)

この機能は試験運用版であり、将来のリリースで更新される可能性があります。

前提条件 (Prerequisites)¶

EventarcToolset を使用する前に、次のセットアップ手順を完了する必要があります。

Eventarc API の有効化: Google Cloud プロジェクトで Eventarc および Eventarc Publishing API を有効にします。
```
gcloud services enable eventarc.googleapis.com eventarcpublishing.googleapis.com
```
認証と権限付与: エージェントを実行するプリンシパルに、Eventarc メッセージバスにメッセージを発行するために必要な IAM 権限（たとえば roles/eventarc.publisher ロール）があることを確認します。Eventarc の IAM ロールの詳細については、Eventarc のアクセス制御ドキュメントを参照してください。ローカル開発用の認証情報を設定するには、アプリケーションデフォルト認証情報 (ADC) の提供を参照してください。
メッセージバスの作成: 発行されたイベントを受信するターゲット Eventarc Advanced メッセージバスを Google Cloud プロジェクトに作成します。
```
gcloud eventarc message-buses create my-bus \
    --location=us-central1 \
    --logging-config=DEBUG
```
必要な依存関係のインストール: 必要な Google Cloud Eventarc クライアントライブラリを含めるために gcp エクストラパッケージをインストールします。
```
pip install "google-adk[gcp]"
```

エージェントでの使用¶

次の例は、CloudEvents を発行するために EventarcToolset を構成し、エージェントに装備する方法を示しています。

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import os

from google.adk.agents import Agent
from google.adk.integrations.eventarc import EventarcCredentialsConfig
from google.adk.integrations.eventarc import EventarcToolConfig
from google.adk.integrations.eventarc import EventarcToolset
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import google.auth

# Define constants for this example agent
AGENT_NAME = "eventarc_agent"
APP_NAME = "eventarc_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-flash-latest"

# Define Eventarc tool config.
# You can optionally set the project_id here, or let the agent infer it from context/user input.
tool_config = EventarcToolConfig(project_id=os.getenv("GOOGLE_CLOUD_PROJECT"))

# Uses externally-managed Application Default Credentials (ADC) by default.
# This decouples authentication from the agent / tool lifecycle.
# https://cloud.google.com/docs/authentication/provide-credentials-adc
application_default_credentials, _ = google.auth.default()
credentials_config = EventarcCredentialsConfig(
    credentials=application_default_credentials
)

# Instantiate an Eventarc toolset
eventarc_toolset = EventarcToolset(
    credentials_config=credentials_config, tool_config=tool_config
)

# Agent Definition
root_agent = Agent(
    model=GEMINI_MODEL,
    name=AGENT_NAME,
    description=(
        "Agent to publish structured CloudEvents to Google Cloud Eventarc"
        " Message Buses."
    ),
    instruction="""\
        You are a cloud integration agent with access to Google Cloud Eventarc tools.
        You can publish structured CloudEvents to Eventarc Message Buses using the publish_message tool.
    """,
    tools=[eventarc_toolset],
)

# Session and Runner
session_service = InMemorySessionService()
session = asyncio.run(
    session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
)
runner = Runner(
    agent=root_agent, app_name=APP_NAME, session_service=session_service
)


# Agent Interaction
def call_agent(query: str):
  """Helper function to call the agent with a query."""
  content = types.Content(role="user", parts=[types.Part(text=query)])
  events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

  print("USER:", query)
  for event in events:
    if event.is_final_response():
      final_response = event.content.parts[0].text
      print("AGENT:", final_response)


# Example call to publish a CloudEvent
call_agent(
    "Publish an event of type 'com.example.user.signup' to bus"
    " 'projects/my-project/locations/us-central1/messageBuses/my-bus' with data"
    " '{\"user\": \"alice\"}', datacontenttype 'application/json', and source"
    " '//my-service/auth'"
)

ツール一覧¶

EventarcToolset には、デフォルトで次の汎用発行ツールが含まれています。

`publish_message`¶

Google Cloud Eventarc Advanced メッセージバスに構造化された CloudEvent を発行します。

パラメータ	型	説明
`bus`	`str`	Eventarc メッセージバスの完全なリソース名（例: `projects/my-project/locations/us-central1/messageBuses/my-bus`）。
`type`	`str`	発生した事象を表す CloudEvents タイプ識別子（例: `com.example.user.signup`）。
`source`	`str`	イベントが発生したコンテキストを識別する CloudEvents ソース URI（例: `//my-service/auth`）。
`data`	`dict \\| str \\| Any`	（オプション）CloudEvent に含めるイベントペイロードデータ。
`datacontenttype`	`str`	（オプション）`data` のメディアタイプ（例: `application/json`）。辞書または JSON データが提供された場合のデフォルトは `application/json` です。
`subject`	`str`	（オプション）イベントプロデューサーのコンテキストにおけるイベントの件名。
`id`	`str`	（オプション）イベントの一意の識別子。指定しない場合、UUID が自動生成されます。
`time`	`str`	（オプション）RFC 3339 形式のイベント発生タイムスタンプ。指定しない場合、現在の UTC タイムスタンプが使用されます。
`specversion`	`str`	（オプション）CloudEvents 仕様バージョン。デフォルトは `1.0` です。
`is_base64_encoded`	`bool`	（オプション）`data` が Base64 エンコードされたバイナリデータかどうか。デフォルトは `False` です。
`include_tracing_extension`	`bool`	（オプション）分散トレーシングコンテキストを自動的に抽出し、CloudEvent の拡張属性に注入するかどうか。デフォルトは `False` です。
`custom_attributes`	`dict[str, str]`	（オプション）イベントに添付する追加のカスタム CloudEvent 拡張属性。

ドメイン固有の発行ツール (Domain-specific publish tools)¶

本番環境のマルチエージェントアーキテクチャにおいて、LLM がルーティングパラメータ（bus、type、source）を自由に設定できるようにすると、ハルシネーションによる誤った送信先や不適切なイベントスキーマが発生する可能性があります。EventarcToolset.create_publish_tool ファクトリメソッドを使用すると、厳格なスキーマを持つドメイン固有の発行ツールを作成できます。

ドメイン固有のツールを作成することで、CloudEventAttributesBinding を使用してルーティング属性をバインドしながら、イベントペイロード（payload_schema）に対して厳格な Pydantic モデルを適用できます。これにより、生成されたイベントがビジネスドメインに一致し、承認されたメッセージバスにのみルーティングされることが保証されます。

エージェントでの使用¶

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import os
from typing import Any

from google.adk.agents import Agent
from google.adk.integrations.eventarc import AgentProvided
from google.adk.integrations.eventarc import CloudEventAttributesBinding
from google.adk.integrations.eventarc import EventarcCredentialsConfig
from google.adk.integrations.eventarc import EventarcToolConfig
from google.adk.integrations.eventarc import EventarcToolset
from google.adk.integrations.eventarc import OMIT
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import google.auth
import pydantic

# Define constants for this example agent
AGENT_NAME = "domain_specific_eventarc_agent"
APP_NAME = "eventarc_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-flash-latest"

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
BUS_NAME = os.getenv("EVENTARC_BUS_NAME", "outreach-bus")
BUS_URI = f"projects/{PROJECT_ID}/locations/us-central1/messageBuses/{BUS_NAME}"


# 1. Define a strictly validated Pydantic schema for the CloudEvent payload
class OutreachContext(pydantic.BaseModel):
  """Structured event payload for a completed customer outreach attempt."""

  customer_id: str = pydantic.Field(
      description="Unique identifier of the customer reached out to."
  )
  resolution_notes: str = pydantic.Field(
      description="Summary notes describing the outcome of the outreach call."
  )
  high_priority: bool = pydantic.Field(
      default=False,
      description="Whether this outreach requires urgent follow-up action.",
  )


# 2. Configure credentials and toolset
tool_config = EventarcToolConfig(project_id=PROJECT_ID)
application_default_credentials, _ = google.auth.default()
credentials_config = EventarcCredentialsConfig(
    credentials=application_default_credentials
)
eventarc_toolset = EventarcToolset(
    credentials_config=credentials_config, tool_config=tool_config
)

# 3. Create Domain-Specific Publish Tools

# Example A: Fully Statically Bound Tool (Safest)
# All routing parameters are locked down by the developer.
# The LLM only provides the structured data matching OutreachContext.
complete_outreach_static_tool = eventarc_toolset.create_publish_tool(
    name="complete_outreach_static",
    description="Logs a completed outreach attempt (statically bound routing).",
    payload_schema=OutreachContext,
    bus=BUS_URI,
    ce_attributes_binding=CloudEventAttributesBinding(
        type="vendor_outreach.completed",
        source="//my-agent/outreach",
        datacontenttype="application/json",
    ),
)

# Example B: Dynamically Bound Tool using AgentProvided and Sentinels
# Allows the LLM to provide the CloudEvent subject, while excluding optional attributes from the event payload.
complete_outreach_dynamic_tool = eventarc_toolset.create_publish_tool(
    name="complete_outreach_dynamic",
    description="Logs an outreach attempt with a dynamically provided subject.",
    payload_schema=OutreachContext,
    bus=BUS_URI,
    ce_attributes_binding=CloudEventAttributesBinding(
        type="vendor_outreach.completed",
        source="//my-agent/outreach",
        subject=AgentProvided("The unique customer ID being reached out to."),
        time=OMIT,
    ),
)


# Example C: Runtime Lambda Binding
# Evaluates attribute values dynamically at execution time from event payload, runtime context, or both.
def resolve_source_from_context(context: Any) -> str:
  """Extracts the source URI dynamically from runtime tool execution context."""
  return f"//my-agent/session/{getattr(context, 'session_id', 'default')}"


complete_outreach_lambda_tool = eventarc_toolset.create_publish_tool(
    name="complete_outreach_lambda",
    description="Logs an outreach attempt using runtime context lambda binding.",
    payload_schema=OutreachContext,
    bus=BUS_URI,
    ce_attributes_binding=CloudEventAttributesBinding(
        type="vendor_outreach.completed",
        source=resolve_source_from_context,
    ),
)

# 4. Equip the agent with the domain-specific tools
root_agent = Agent(
    model=GEMINI_MODEL,
    name=AGENT_NAME,
    description="Agent for recording customer outreach completion events.",
    instruction="""\
        You are a customer outreach agent.
        Use the available outreach tools to record structured outreach events.
    """,
    tools=[
        complete_outreach_static_tool,
        complete_outreach_dynamic_tool,
        complete_outreach_lambda_tool,
    ],
)

# 5. Session and Runner
session_service = InMemorySessionService()
session = asyncio.run(
    session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
)
runner = Runner(
    agent=root_agent, app_name=APP_NAME, session_service=session_service
)


def call_agent(query: str):
  """Helper function to call the agent with a query."""
  content = types.Content(role="user", parts=[types.Part(text=query)])
  events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

  print("USER:", query)
  for event in events:
    if event.is_final_response():
      final_response = event.content.parts[0].text
      print("AGENT:", final_response)


# Example invocation
call_agent(
    "We successfully completed an outreach call with CUST-883. "
    "Resolution notes: All issues resolved. Not high priority."
)

`create_publish_tool` のパラメータ¶

create_publish_tool メソッドは、以下のキーワード専用引数を受け入れます。

パラメータ	型	説明
`name`	`str`	LLM に公開される関数ツール名（例: `complete_outreach_static`）。
`description`	`str`	LLM に対してこのツールをいつ呼び出すか、どのようなアクションを実行するかを指示する自然言語による説明。
`bus`	`str \\| Callable[[Any], str] \\| AgentProvided`	ターゲットとなる Eventarc メッセージバス。固定の URI 文字列、ツールのコンテキストで評価されるランタイム Callable、または LLM に供給を求める `AgentProvided` インスタンスが可能です。
`ce_attributes_binding`	`CloudEventAttributesBinding`	CloudEvent 属性（`type`、`source`、`subject`、`datacontenttype`、`time`、`id`、`specversion`、`custom_attributes`）のバインディングルール。
`payload_schema`	`type[pydantic.BaseModel] \\| None`	（オプション）構造化イベントペイロードを定義する Pydantic スキーマクラス。指定された場合、ツールシグネチャにはこのモデルに準拠する `event_data` パラメータが必要となります。指定されない場合（または `None` の場合）、ツールシグネチャに `event_data` パラメータは追加されず、ツールはデータペイロード本体のない通知専用 CloudEvent を発行します。

CloudEvent 属性のバインディングとセンチネル (Attribute bindings and sentinels)¶

CloudEventAttributesBinding データクラスは、個々の CloudEvent フィールドがどのように設定されるかを構成します。各属性（type、source、datacontenttype、subject、time、id、specversion、custom_attributes）には、以下のバインディングメカニズムのいずれかを割り当てることができます。

バインディングタイプ	例	LLMへの公開	説明
Static String	`type="vendor_outreach.completed"`	いいえ	固定のリテラル文字列を適用します。この属性は LLM シグネチャから隠され、毎回の呼び出しに自動的に適用されます。
Runtime Lambda	`source=lambda ctx: f"//agent/{ctx.id}"`	いいえ	ツールのランタイムコンテキストを使用して実行時に動的に評価される Callable (`Callable[[Any], str]`) です。LLM シグネチャから隠されます。
`AgentProvided`	`subject=AgentProvided("Customer ID")`	はい	ADK に対し、属性を関数シグネチャの明示的なパラメータとして公開し、LLM が提供できるように指示します。`description` 文字列を受け入れます。
`MISSING`	`time=MISSING`	いいえ	オプション属性のデフォルトセンチネルです。デフォルトの動作が適用されることを示します（例: `time` の場合は現在の UTC タイムスタンプを自動生成、`id` の場合は UUID を自動生成）。
`OMIT`	`time=OMIT`	いいえ	生成された CloudEvent からオプション属性を明示的に除外します。必須属性（`type`、`source`、`bus`）を `OMIT` に設定することはできません。

例: `MISSING` と `OMIT` の違いの理解¶

MISSING と OMIT の違いを理解するために、time などのオプションの CloudEvent 属性にどのように影響するかを見てみましょう。

time=MISSING（デフォルトの動作）: time=MISSING に設定する（または time を未指定のままにする）と、ツールセットはその組み込みのデフォルト動作を適用します。time の場合、RFC 3339 形式で現在の UTC タイムスタンプを自動的に生成して含めます（例: "time": "2026-07-31T20:20:00Z"）。
time=OMIT: 明示的に time=OMIT に設定すると、発行された CloudEvent ペイロードから time フィールドが完全に除外されます。ダウンストリームのイベントコンシューマーがオプションの属性を必要としない、または想定していない場合は OMIT を使用します。

from google.adk.integrations.eventarc import (
    CloudEventAttributesBinding,
    MISSING,
    OMIT,
)

# 1. MISSING の使用 (デフォルト): CloudEvent に現在の UTC タイムスタンプが自動的に含まれます
binding_with_timestamp = CloudEventAttributesBinding(
    type="vendor_outreach.completed",
    source="//my-agent/outreach",
    time=MISSING,  # 結果: "time": "2026-07-31T20:20:00Z"
)

# 2. OMIT の使用: CloudEvent に 'time' 属性が含まれません
binding_without_timestamp = CloudEventAttributesBinding(
    type="vendor_outreach.completed",
    source="//my-agent/outreach",
    time=OMIT,  # 発行されたイベントから 'time' フィールドが除外されます
)

ADK 用 Google Cloud Eventarc ツール¶

前提条件 (Prerequisites)¶

エージェントでの使用¶

ツール一覧¶

publish_message¶

ドメイン固有の発行ツール (Domain-specific publish tools)¶

エージェントでの使用¶

create_publish_tool のパラメータ¶

CloudEvent 属性のバインディングとセンチネル (Attribute bindings and sentinels)¶

例: MISSING と OMIT の違いの理解¶

追加リソース¶

`publish_message`¶

`create_publish_tool` のパラメータ¶

例: `MISSING` と `OMIT` の違いの理解¶