How can I add custom_metadata to an event

[SamiAlsubhi]

I noticed there is a column custom_metadata in table events. How can I add information to it. I want to use it to filter context based on topic change. I created a plugin for this. However, now I am not sure how to manipulate this custom_metadata. Could please explain to me how I can custom metadata to an event.

from __future__ import annotations

import logging
from typing import List, Optional

from google.adk.agents.callback_context import CallbackContext
from google.adk.events.event import Event
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.plugins import BasePlugin
from google.genai import types

logger = logging.getLogger(__name__)


class ContextTopicFilterPlugin(BasePlugin):
    """A plugin that filters LLM context based on topic changes in user events.

    Scans events from latest to oldest, checking user events for topic metadata.
    When the topic value changes, truncates all older events to maintain
    topic-based conversation context.

    Note: If the latest user event has no custom_metadata or the topic key doesn't
    exist, the plugin will keep all events where user messages also have no topic
    metadata (graceful degradation behavior).
    """

    def __init__(
        self,
        event_metadata_topic_key: str = "topic",
        name: str = "context_topic_filter_plugin",
    ):
        """Initializes the context topic filter plugin.

        Args:
          event_metadata_topic_key: The key to look for in user event custom_metadata.
            Defaults to "topic".
          name: The name of the plugin instance.
        """
        super().__init__(name)
        self._event_metadata_topic_key = event_metadata_topic_key

    async def before_model_callback(
        self, *, callback_context: CallbackContext, llm_request: LlmRequest
    ) -> Optional[LlmResponse]:
        """Filters LLM context to only include events from the current topic."""
        try:
            events = callback_context.session.events
            latest_topic = self._get_latest_topic_from_user_events(events)
            llm_request.contents = self._truncate_events_to_latest_topic(
                latest_topic, events
            )
        except Exception as e:
            logger.error(f"Failed to reduce context for request: {e}")

        return None

    def _get_latest_topic_from_user_events(self, events: List[Event]) -> Optional[str]:
        """Gets the topic value from the most recent user event."""
        # Filter to only user events with content
        events_with_content = [
            event for event in events if event.content and event.author == "user"
        ]

        # Handle edge case: no user events with content
        if not events_with_content:
            return None

        # Get the latest topic value from the most recent user event
        latest_topic = None
        if metadata := events_with_content[-1].custom_metadata:
            latest_topic = metadata.get(self._event_metadata_topic_key, None)

        return latest_topic

    def _truncate_events_to_latest_topic(
        self, latest_topic: str | None, events: List[Event]
    ) -> List[types.Content]:
        """Returns event contents from the current topic onwards."""
        if not events:
            return []

        # Get the latest topic value from user events

        # If no topic found, return all event contents
        if latest_topic is None:
            return [event.content for event in events if event.content]

        # Find truncation point by scanning from latest to oldest
        truncate_index = 0
        for i in range(len(events) - 1, -1, -1):
            event = events[i]

            # Only check topic in user events
            if event.content is None or event.author != "user":
                continue

            current_topic = None
            if event.custom_metadata:
                current_topic = event.custom_metadata.get(
                    self._event_metadata_topic_key, None
                )

            # Check if topic matches (including both being None)
            if current_topic == latest_topic:
                truncate_index = i
            else:
                # Topic changed - stop here
                break

        # Return content from all events after truncation point
        return [event.content for event in events[truncate_index:] if event.content]

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: You can add custom_metadata to events by passing a RunConfig object with your desired metadata to the agent runner's run method.

---

Hello! You can add custom metadata to all events generated within a single invocation of your agent by using the RunConfig object. This object is passed to the agent runner's run method (e.g., run_live or run_async).

The metadata you provide in RunConfig will be attached to the custom_metadata field of each Event created during that run. [1]

Here is an example of how you can do it:

from google.adk.agents.run_config import RunConfig

# Define your custom metadata
run_config = RunConfig(
    custom_metadata={
        "topic": "your_topic_value",
        "some_other_key": "some_other_value",
    }
)

# Pass the config when running the agent
# This assumes you have a 'runner' instance of AgentRunner
await runner.run_live(
    user_input="Some user message",
    run_config=run_config
)

By setting the "topic" in the custom_metadata for each run call, your ContextTopicFilterPlugin will then be able to access it from the Event objects and filter the context as you intended.

---

[1] https://google.github.io/adk-docs/streaming/dev-guide/part4/

[SamiAlsubhi]

I did that but it was not reflected in the db in events.custom_metadata!

        run_async_generator = self.runner.run_async(
            user_id=request.user_id,
            session_id=session_id,
            invocation_id=invocation_id,
            new_message=types.Content(role="user", parts=[part]),
            run_config=RunConfig(custom_metadata={"topic": "example_topic"}),
        )

[SamiAlsubhi]

ok it looks like there is already and issue raised about this and a fix is already committed.

#3953