feat(eventarc): Support Context callables and correct OMIT behavior

Merge #6600

**1\. Link to an existing issue (if applicable):**

- Related: google/adk-docs\#2045 (Addresses technical verification feedback in [https://github.com/google/adk-docs/pull/2045\#issuecomment-5184438669](https://github.com/google/adk-docs/pull/2045#issuecomment-5184438669))

**2\. Or, if no issue exists, describe the change:**

**Problem:**

1. Callable attribute bindings in `CloudEventAttributesBinding` were always evaluated against the event `payload`. This prevented developers from correlating CloudEvents with ADK runtime telemetry (such as session IDs or invocation IDs from `Context`).
2. Setting `time=OMIT` or `datacontenttype=OMIT` in `CloudEventAttributesBinding` skipped adding keyword arguments when calling `publish_message`. Because `publish_message` auto-generates default UTC timestamps and content types when arguments are `None` or omitted, `time=OMIT` generated a timestamp instead of omitting the header.
3. Sample READMEs omitted the required `pip install "google-adk[gcp]"` prerequisite step needed for Eventarc publishing.

**Solution:**

1. Added automatic signature inspection (`0-arg`, `1-arg`, and `2-arg` callables) to `CloudEventAttributesBinding` so callables can receive the event `payload`, the runtime `Context` (`tool_context`), or both, while preserving full backward compatibility with existing payload callbacks.
2. Setting `time=OMIT` or `datacontenttype=OMIT` now explicitly passes empty string (`""`) to `publish_message` so attributes are omitted from published CloudEvents. Explicitly setting required CloudEvent specification headers (`id=OMIT`, `specversion=OMIT`) now raises a `TypeError` at tool build time.
3. Updated sample agents and sample READMEs (`domain_specific_agent` and `generic_agent`) to demonstrate `Context` callables, `time=OMIT`, and GCP extra prerequisites.

### Testing Plan

**Unit Tests:**

- I have added or updated unit tests for my change.
- All unit tests pass locally.

Summary of passed `pytest` results:

```
uv run --all-extras pytest tests/unittests/integrations/eventarc -v

======================== 63 passed, 4 warnings, 13 subtests passed in 3.14s ========================
```

- Added `test_runtime_execution_with_context_and_payload_lambdas` to verify 1-parameter (`payload` or `Context`) and 2-parameter callables.
- Added `test_time_and_datacontenttype_omit_pass_empty_string` to verify omission of `time` and `datacontenttype`.
- Added `test_id_and_specversion_omit_raise_typeerror` to verify static validation against omitting mandatory CloudEvent specification headers.

**Manual End-to-End (E2E) Tests:**

- Verified that sample agent tools in `contributing/samples/integrations/eventarc/domain_specific_agent/agent.py` build and run correctly.
- Confirmed that `complete_outreach_lambda_tool` correctly injects `Context.session_id` into the event source and that `ping_system_tool` emits events without a timestamp header when configured with `time=OMIT`.

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document.
- [x] I have performed a self-review of my own code.
- [x] I have commented my code, particularly in hard-to-understand areas.
- [x] I have added tests that prove my fix is effective or that my feature works.
- [x] New and existing unit tests pass locally with my changes.
- [x] I have manually tested my changes end-to-end.
- [x] Any dependent changes have been merged and published in downstream modules.

Additional context

Addresses the technical verification report on google/adk-docs#2045 comment.

COPYBARA_INTEGRATE_REVIEW=#6600 from milenvk:main 647744f
PiperOrigin-RevId: 964084439

Author: milenvk

contributing/samples/integrations/eventarc/domain_specific_agent/README.md

@@ -46,6 +46,12 @@ gcloud eventarc message-buses create my-bus \
 
 *(Make sure to update the `BUS_NAME` variable in `agent.py` to match your actual bus URI).*
 
+3. Install the GCP extra dependency (required for Eventarc publishing):
+
+```bash
+pip install "google-adk[gcp]"
+```
+
 `create_publish_tool` is highly flexible. It uses `pydantic.create_model` to construct the LLM's function signature, encapsulating the `payload_schema` inside an `event_data` parameter and appending any parameter marked with `AgentProvided`.
 
 ### Example A: Fully Statically Bound (Safest)
@@ -89,21 +95,24 @@ complete_outreach_dynamic_tool = toolset.create_publish_tool(
 
 ### Example C: Lambda Execution & Mixed Custom Attributes
 
-The developer uses Python callables to generate IDs dynamically at runtime.
+The developer uses Python callables to generate attributes dynamically at runtime. Callables can inspect the event payload, the agent's runtime `Context` (`tool_context`), or both.
 
 ```python
 def get_custom_trace_id(payload: OutreachContext) -> str:
     return f"trace-{payload.customer_id}-{uuid.uuid4().hex[:8]}"
 
+def get_source_from_session(ctx: Context) -> str:
+    return f"//my-agent/outreach/{ctx.session_id}"
+
 complete_outreach_lambda_tool = toolset.create_publish_tool(
     name="complete_outreach_lambda",
     description="Logs a completed outreach attempt.",
     payload_schema=OutreachContext,
     bus=f"projects/{PROJECT_ID}/locations/us-central1/messageBuses/{BUS_NAME}",
     ce_attributes_binding=CloudEventAttributesBinding(
         type="vendor_outreach.completed",
-        source="//my-agent/outreach",
-        id=get_custom_trace_id, # Executed at runtime
+        source=get_source_from_session, # Evaluated against runtime Context
+        id=get_custom_trace_id, # Evaluated against event payload
         custom_attributes={
             "environment": "production", # Statically bound
             "priority": AgentProvided("The priority of the outreach: 'high' or 'low'")
@@ -114,9 +123,9 @@ complete_outreach_lambda_tool = toolset.create_publish_tool(
 
 **What the Agent Sees:** `complete_outreach_lambda(event_data: OutreachContext, priority: str)`
 
-### Example D: Empty Payloads & Dynamic Defaults
+### Example D: Empty Payloads, Omit Headers & Dynamic Defaults
 
-The developer wants to emit a simple signal (no business payload). If the agent omits the priority, it is dynamically calculated.
+The developer wants to emit a simple signal (no business payload) without a timestamp header (`time=OMIT`). If the agent omits the priority, it is dynamically calculated.
 
 ```python
 def default_priority(_: None) -> str:
@@ -130,6 +139,7 @@ ping_system_tool = toolset.create_publish_tool(
     ce_attributes_binding=CloudEventAttributesBinding(
         type="system.ping",
         source="//my-agent/ping",
+        time=OMIT, # Omits time attribute from event
         custom_attributes={
             "retry": AgentProvided("Whether to retry on failure", default="false"),
             "priority": AgentProvided("The priority of the ping", default=default_priority)

contributing/samples/integrations/eventarc/domain_specific_agent/agent.py

@@ -17,6 +17,7 @@
 import uuid
 
 from google.adk.agents import llm_agent
+from google.adk.agents.context import Context
 from google.adk.auth import auth_credential
 from google.adk.integrations.eventarc import AgentProvided
 from google.adk.integrations.eventarc import CloudEventAttributesBinding
@@ -104,11 +105,16 @@ class OutreachContext(pydantic.BaseModel):
 
 
 # Example C: Lambda Execution & Mixed Custom Attributes
-# The developer uses Python callables to generate IDs dynamically at runtime.
+# The developer uses Python callables to generate attributes dynamically at runtime.
+# Callables can inspect the event payload, the runtime Context, or both.
 def get_custom_trace_id(payload: OutreachContext) -> str:
   return f"trace-{payload.customer_id}-{uuid.uuid4().hex[:8]}"
 
 
+def get_source_from_session(ctx: Context) -> str:
+  return f"//my-agent/outreach/{ctx.session_id}"
+
+
 complete_outreach_lambda_tool = toolset.create_publish_tool(
     name="complete_outreach_lambda",
     description=(
@@ -119,7 +125,7 @@ def get_custom_trace_id(payload: OutreachContext) -> str:
     bus=f"projects/{PROJECT_ID}/locations/us-central1/messageBuses/{BUS_NAME}",
     ce_attributes_binding=CloudEventAttributesBinding(
         type="vendor_outreach.completed",
-        source="//my-agent/outreach",
+        source=get_source_from_session,
         id=get_custom_trace_id,
         custom_attributes={
             "environment": "production",
@@ -145,6 +151,7 @@ def default_priority(_: None) -> str:
     ce_attributes_binding=CloudEventAttributesBinding(
         type="system.ping",
         source="//my-agent/ping",
+        time=OMIT,  # Omits time attribute from event
         custom_attributes={
             "retry": AgentProvided(
                 "Whether to retry on failure", default="false"

contributing/samples/integrations/eventarc/generic_agent/README.md

@@ -39,6 +39,12 @@ gcloud eventarc message-buses create my-bus \
 
 *(Make sure to update the `BUS_NAME` variable in `agent.py` to match your actual bus URI).*
 
+3. Install the GCP extra dependency (required for Eventarc publishing):
+
+```bash
+pip install "google-adk[gcp]"
+```
+
 Set up environment variables in your `.env` file for using Google AI Studio or Google Cloud Vertex AI for the LLM service. For example:
 
 - `GOOGLE_GENAI_USE_ENTERPRISE=FALSE`

src/google/adk/integrations/eventarc/_domain_specific_publish.py

@@ -24,6 +24,7 @@
 
 from google.adk.agents.context import Context
 from google.adk.tools.google_tool import GoogleTool
+from google.adk.utils.context_utils import find_context_parameter
 import google.auth.credentials
 import pydantic
 
@@ -45,6 +46,10 @@ class OmitSentinel:
 OMIT = OmitSentinel()
 
 
+def _is_context_param(func: Any, param_name: str) -> bool:
+  return find_context_parameter(func) == param_name
+
+
 @dataclass
 class AgentProvided:
   """Indicates that a CloudEvent attribute should be provided by the LLM."""
@@ -53,26 +58,37 @@ class AgentProvided:
   default: Any | OmitSentinel | MissingSentinel = MISSING
 
 
-AttributeBinding = str | Callable[[Any], str] | AgentProvided
+AttributeBinding = str | Callable[..., str] | AgentProvided
 OptionalAttributeBinding = (
     str
-    | Callable[[Any], str | OmitSentinel]
+    | Callable[..., str | OmitSentinel]
     | AgentProvided
     | OmitSentinel
     | MissingSentinel
     | None
 )
 CustomAttributeBinding = (
-    str | Callable[[Any], str | OmitSentinel] | AgentProvided | OmitSentinel
+    str | Callable[..., str | OmitSentinel] | AgentProvided | OmitSentinel
 )
 SpecVersionBinding = (
-    str | Callable[[Any], str] | AgentProvided | MissingSentinel | None
+    str | Callable[..., str] | AgentProvided | MissingSentinel | None
 )
 
 
 @dataclass
 class CloudEventAttributesBinding:
-  """Configuration for binding CloudEvent attributes to static values, lambdas, or AgentProvided fields."""
+  """Configuration for binding CloudEvent attributes to static values, lambdas, or AgentProvided fields.
+
+  Lambda/callable bindings can accept:
+  - 1 parameter for the event payload (`lambda p: ...`)
+  - 1 parameter for the runtime context (`lambda ctx: ...` or type-annotated with `Context`)
+  - 2 parameters for both (`lambda p, ctx: ...`)
+  - 0 parameters (`lambda: ...`)
+
+  Setting optional attributes (`time`, `datacontenttype`, `subject`,
+  `custom_attributes`) to `OMIT` omits them from the published CloudEvent.
+  Required attributes (`type`, `source`, `id`, `specversion`) cannot be `OMIT`.
+  """
 
   type: AttributeBinding
   source: AttributeBinding
@@ -92,17 +108,21 @@ def build_domain_specific_tool(
     ce_attributes_binding: CloudEventAttributesBinding,
     payload_schema: type[pydantic.BaseModel] | None = None,
 ) -> GoogleTool:
-  """Dynamically builds a GoogleTool wrapping publish_message with specific bindings."""
+  """Dynamically builds a GoogleTool wrapping publish_message with specific bindings.
+
+  Callable bindings in `ce_attributes_binding` can inspect the event payload, the
+  runtime `Context` (`tool_context`), or both.
+  """
 
   # 1. Validation
   mandatory_fields = ["type", "source"]
   for field in mandatory_fields:
     val = getattr(ce_attributes_binding, field)
-    if val is MISSING:
+    if val is MISSING:  # type: ignore[comparison-overlap]
       raise TypeError(
           f"CloudEventAttributesBinding requires '{field}' to be provided."
       )
-    if val is OMIT:
+    if val is OMIT:  # type: ignore[comparison-overlap]
       raise TypeError(
           f"CloudEvent field '{field}' is mandatory and cannot be OMIT."
       )
@@ -118,6 +138,13 @@ def build_domain_specific_tool(
   if bus is None:
     raise TypeError("The 'bus' parameter is mandatory and cannot be None.")
 
+  for field in ("id", "specversion"):
+    val = getattr(ce_attributes_binding, field)
+    if val is OMIT:  # type: ignore[comparison-overlap]
+      raise TypeError(
+          f"CloudEvent field '{field}' is mandatory and cannot be OMIT."
+      )
+
   reserved_attributes = {
       "type",
       "source",
@@ -300,7 +327,29 @@ def resolve_attr(key: str, binding: Any, is_mandatory: bool) -> Any:
 
       # Evaluate lambdas
       if callable(val):
-        val = val(payload)
+        tool_context = kwargs.get("tool_context")
+        try:
+          sig = inspect.signature(val)
+        except (ValueError, TypeError):
+          sig = None
+
+        if sig is not None:
+          params = list(sig.parameters.values())
+          if len(params) == 2:
+            first_param = params[0]
+            if _is_context_param(val, first_param.name):
+              val = val(tool_context, payload)
+            else:
+              val = val(payload, tool_context)
+          elif len(params) == 1:
+            if _is_context_param(val, params[0].name):
+              val = val(tool_context)
+            else:
+              val = val(payload)
+          else:
+            val = val()
+        else:
+          val = val(payload)
 
       if val is OMIT:
         if is_mandatory:
@@ -325,7 +374,14 @@ def resolve_attr(key: str, binding: Any, is_mandatory: bool) -> Any:
       val = resolve_attr(
           field, getattr(ce_attributes_binding, field), is_mandatory
       )
-      if val is not OMIT and val is not None:
+      if val is OMIT:
+        if field in ("time", "datacontenttype"):
+          publish_kwargs[field] = ""
+        elif field in ("id", "specversion"):
+          raise ValueError(
+              f"CloudEvent attribute '{field}' is mandatory and cannot be OMIT."
+          )
+      elif val is not None:
         publish_kwargs[field] = val
 
     # Resolve custom attributes