Google Cloud Eventarc tool for ADK¶

Supported in ADKPython v2.6.0Experimental

EventarcToolset 允许智能体与 Google Cloud Eventarc 交互，异步发布结构化的 CloudEvents 到 Eventarc 消息总线。该工具集在多次调用之间提供内置的连接池和缓存，并且支持通用事件发布和领域特定的、带 schema 校验的事件工具。

实验性功能

此功能为实验性功能，可能会在后续版本中更新。

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 访问控制文档。要设置本地开发凭据，请参阅提供应用默认凭据。
创建消息总线：在你的 Google Cloud 项目中创建一个目标 Eventarc 高级消息总线，用于接收发布的事件：
```
gcloud eventarc message-buses create my-bus \
    --location=us-central1 \
    --logging-config=DEBUG
```

Use with agent¶

以下示例展示了如何配置并使用 EventarcToolset 来为智能体发布 CloudEvents：

# 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\"}' and source '//my-service/auth'"
)

Tools¶

EventarcToolset 默认包含以下通用发布工具：

`publish_message`¶

向 Google Cloud Eventarc 高级消息总线发布结构化的 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）可能导致虚构的目标地址或格式错误的事件 schema。EventarcToolset.create_publish_tool 工厂方法允许你创建领域特定的、严格 schema 的发布工具。

通过创建领域特定的工具，你可以使用 CloudEventAttributesBinding 绑定路由属性，同时强制要求事件负载（payload_schema）遵循严格的 Pydantic 模型。这保证了生成的事件匹配你的业务领域，并且仅路由到授权的消息总线。

Use with agent¶

# 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 runtime context.
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."
)

Parameters for `create_publish_tool`¶

create_publish_tool 方法接受以下关键字参数：

参数	类型	描述
`name`	`str`	暴露给 LLM 的函数工具名称（例如 `complete_outreach_static`）。
`description`	`str`	用于指导 LLM 何时调用此工具以及该工具执行什么操作的自然语言描述。
`bus`	`str \\| Callable[[Any], str] \\| AgentProvided`	目标 Eventarc 消息总线。可以是静态 URI 字符串、在运行时根据工具上下文求值的可调用对象，或一个 `AgentProvided` 实例以提示 LLM 提供该值。
`ce_attributes_binding`	`CloudEventAttributesBinding`	CloudEvent 属性（`type`、`source`、`subject`、`datacontenttype`、`time`、`id`、`specversion`、`custom_attributes`）的绑定规则。
`payload_schema`	`type[pydantic.BaseModel] \\| None`	（可选）定义结构化事件负载的 Pydantic schema 类。指定后，工具签名将要求一个符合此模型的 `event_data` 参数。如果未提供（或为 `None`），则不会在工具签名中添加 `event_data` 参数，工具将发布一个仅通知的、不带数据负载体的 CloudEvent。

CloudEvent attribute bindings and sentinels¶

CloudEventAttributesBinding 数据类用于配置各个 CloudEvent 字段的填充方式。每个属性（type、source、datacontenttype、subject、time、id、specversion、custom_attributes）可以分配以下绑定机制之一：

绑定类型	示例	是否暴露给 LLM	描述
静态字符串	`type="vendor_outreach.completed"`	否	强制使用固定的字面字符串。该属性对 LLM 签名隐藏，每次调用时自动应用。
运行时 Lambda	`source=lambda ctx: f"//agent/{ctx.id}"`	否	在执行时使用工具运行时上下文动态求值的可调用对象（`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`。

Example: Understanding `MISSING` versus `OMIT`¶

要理解 MISSING 和 OMIT 之间的区别，请考虑它们如何影响可选的 CloudEvent 属性（如 time）：

time=MISSING（默认行为）：当你设置 time=MISSING（或不指定 time）时，工具集将应用其内置的默认行为。对于 time，它会自动生成并包含格式为 RFC 3339 的当前 UTC 时间戳（例如 "time": "2026-07-31T20:20:00Z"）。
time=OMIT：当你明确设置 time=OMIT 时，time 字段将从发布的 CloudEvent 负载中完全排除。当下游事件消费者不需要或不期望可选属性时，使用 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' 字段从发布的事件中排除
)

Google Cloud Eventarc tool for ADK¶

Prerequisites¶

Use with agent¶

Tools¶

publish_message¶

Domain-specific publish tools¶

Use with agent¶

Parameters for create_publish_tool¶

CloudEvent attribute bindings and sentinels¶

Example: Understanding MISSING versus OMIT¶

Additional resources¶

`publish_message`¶

Parameters for `create_publish_tool`¶

Example: Understanding `MISSING` versus `OMIT`¶