feat: Allow list of events to be passed to AdkApp when querying

PiperOrigin-RevId: 844834249

Author: yeesian

vertexai/agent_engines/templates/adk.py

@@ -932,6 +932,7 @@ async def async_stream_query(
         message: Union[str, Dict[str, Any]],
         user_id: str,
         session_id: Optional[str] = None,
+        session_events: Optional[Dict[str, Any]] = None,
         run_config: Optional[Dict[str, Any]] = None,
         **kwargs,
     ) -> AsyncIterable[Dict[str, Any]]:
@@ -944,7 +945,11 @@ async def async_stream_query(
                 Required. The ID of the user.
             session_id (str):
                 Optional. The ID of the session. If not provided, a new
-                session will be created for the user.
+                session will be created for the user. If this is specified, then
+                `session_events` will be ignored.
+            session_events (Optional[List[Dict[str, Any]]]):
+                Optional. The session events to use for the query. This will be
+                used to initialize the session if `session_id` is not provided.
             run_config (Optional[Dict[str, Any]]):
                 Optional. The run config to use for the query. If you want to
                 pass in a `run_config` pydantic object, you can pass in a dict
@@ -974,6 +979,18 @@ async def async_stream_query(
         if not session_id:
             session = await self.async_create_session(user_id=user_id)
             session_id = session.id
+            if session_events is not None:
+                # We allow for session_events to be an empty list.
+                from google.adk.events.event import Event
+
+                session_service = self._tmpl_attrs.get("session_service")
+                async for event in session_events:
+                    if not isinstance(event, Event):
+                        event = Event.model_validate(event)
+                    await session_service.append_event(
+                        session=session,
+                        event=event,
+                    )
 
         run_config = _validate_run_config(run_config)
         if run_config:
@@ -1009,6 +1026,7 @@ def stream_query(
         message: Union[str, Dict[str, Any]],
         user_id: str,
         session_id: Optional[str] = None,
+        session_events: Optional[Dict[str, Any]] = None,
         run_config: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
@@ -1023,7 +1041,11 @@ def stream_query(
                 Required. The ID of the user.
             session_id (str):
                 Optional. The ID of the session. If not provided, a new
-                session will be created for the user.
+                session will be created for the user. If this is specified, then
+                `session_events` will be ignored.
+            session_events (Optional[Dict[str, Any]]):
+                Optional. The session events to use for the query. This will be
+                used to initialize the session if `session_id` is not provided.
             run_config (Optional[Dict[str, Any]]):
                 Optional. The run config to use for the query. If you want to
                 pass in a `run_config` pydantic object, you can pass in a dict
@@ -1063,6 +1085,16 @@ def stream_query(
         if not session_id:
             session = self.create_session(user_id=user_id)
             session_id = session.id
+            if session_events is not None:
+                # We allow for session_events to be an empty list.
+                from google.adk.events.event import Event
+
+                session_service = self._tmpl_attrs.get("session_service")
+                for event_dict in session_events:
+                    await session_service.append_event(
+                        session=session,
+                        event=Event.model_validate(event_dict),
+                    )
         run_config = _validate_run_config(run_config)
         if run_config:
             for event in self._tmpl_attrs.get("runner").run(