feat(runner): add metadata parameter to Runner.run_async()

src/google/adk/agents/invocation_context.py

@@ -206,6 +206,15 @@ class InvocationContext(BaseModel):
   canonical_tools_cache: Optional[list[BaseTool]] = None
   """The cache of canonical tools for this invocation."""
 
+  metadata: Optional[dict[str, Any]] = None
+  """Per-request metadata passed from Runner.run_async().
+
+  This field allows passing arbitrary metadata that can be accessed during
+  the invocation lifecycle, particularly in callbacks like before_model_callback.
+  Common use cases include passing user_id, trace_id, memory context keys, or
+  other request-specific context that needs to be available during processing.
+  """
+
   _invocation_cost_manager: _InvocationCostManager = PrivateAttr(
       default_factory=_InvocationCostManager
   )

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -89,7 +89,7 @@ async def run_live(
       invocation_context: InvocationContext,
   ) -> AsyncGenerator[Event, None]:
     """Runs the flow using live api."""
-    llm_request = LlmRequest()
+    llm_request = LlmRequest(metadata=invocation_context.metadata)
     event_id = Event.new_id()
 
     # Preprocess before calling the LLM.
@@ -380,7 +380,7 @@ async def _run_one_step_async(
       invocation_context: InvocationContext,
   ) -> AsyncGenerator[Event, None]:
     """One step means one LLM call."""
-    llm_request = LlmRequest()
+    llm_request = LlmRequest(metadata=invocation_context.metadata)
 
     # Preprocess before calling the LLM.
     async with Aclosing(

src/google/adk/models/llm_request.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 from typing import Optional
 from typing import Union
 
@@ -99,6 +100,15 @@ class LlmRequest(BaseModel):
   the full history.
   """
 
+  metadata: Optional[dict[str, Any]] = None
+  """Per-request metadata for callbacks and custom processing.
+
+  This field allows passing arbitrary metadata from the Runner.run_async()
+  call to callbacks like before_model_callback. This is useful for passing
+  request-specific context such as user_id, trace_id, or memory context keys
+  that need to be available during model invocation.
+  """
+
   def append_instructions(
       self, instructions: Union[list[str], types.Content]
   ) -> list[types.Content]:

src/google/adk/runners.py

@@ -400,6 +400,7 @@ async def run_async(
       new_message: Optional[types.Content] = None,
       state_delta: Optional[dict[str, Any]] = None,
       run_config: Optional[RunConfig] = None,
+      metadata: Optional[dict[str, Any]] = None,
   ) -> AsyncGenerator[Event, None]:
     """Main entry method to run the agent in this runner.
 
@@ -417,6 +418,13 @@ async def run_async(
       new_message: A new message to append to the session.
       state_delta: Optional state changes to apply to the session.
       run_config: The run config for the agent.
+      metadata: Optional per-request metadata that will be passed to callbacks.
+        This allows passing request-specific context such as user_id, trace_id,
+        or memory context keys to before_model_callback and other callbacks.
+        Note: A shallow copy is made of this dictionary, so top-level changes
+        within callbacks won't affect the original. However, modifications to
+        nested mutable objects (e.g., nested dicts or lists) will affect the
+        original.
 
     Yields:
       The events generated by the agent.
@@ -426,13 +434,16 @@ async def run_async(
         new_message are None.
     """
     run_config = run_config or RunConfig()
+    # Create a shallow copy to isolate from caller's modifications
+    metadata = metadata.copy() if metadata else None
 
     if new_message and not new_message.role:
       new_message.role = 'user'
 
     async def _run_with_trace(
         new_message: Optional[types.Content] = None,
         invocation_id: Optional[str] = None,
+        metadata: Optional[dict[str, Any]] = None,
     ) -> AsyncGenerator[Event, None]:
       with tracer.start_as_current_span('invocation'):
         session = await self.session_service.get_session(
@@ -463,6 +474,7 @@ async def _run_with_trace(
               invocation_id=invocation_id,
               run_config=run_config,
               state_delta=state_delta,
+              metadata=metadata,
           )
           if invocation_context.end_of_agents.get(
               invocation_context.agent.name
@@ -476,6 +488,7 @@ async def _run_with_trace(
               new_message=new_message,  # new_message is not None.
               run_config=run_config,
               state_delta=state_delta,
+              metadata=metadata,
           )
 
         async def execute(ctx: InvocationContext) -> AsyncGenerator[Event]:
@@ -502,7 +515,9 @@ async def execute(ctx: InvocationContext) -> AsyncGenerator[Event]:
               self.app, session, self.session_service
           )
 
-    async with Aclosing(_run_with_trace(new_message, invocation_id)) as agen:
+    async with Aclosing(
+        _run_with_trace(new_message, invocation_id, metadata)
+    ) as agen:
       async for event in agen:
         yield event
 
@@ -1186,6 +1201,7 @@ async def _setup_context_for_new_invocation(
       new_message: types.Content,
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
+      metadata: Optional[dict[str, Any]] = None,
   ) -> InvocationContext:
     """Sets up the context for a new invocation.
 
@@ -1194,6 +1210,7 @@ async def _setup_context_for_new_invocation(
       new_message: The new message to process and append to the session.
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
+      metadata: Optional per-request metadata to pass to callbacks.
 
     Returns:
       The invocation context for the new invocation.
@@ -1203,6 +1220,7 @@ async def _setup_context_for_new_invocation(
         session,
         new_message=new_message,
         run_config=run_config,
+        metadata=metadata,
     )
     # Step 2: Handle new message, by running callbacks and appending to
     # session.
@@ -1225,6 +1243,7 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: Optional[str],
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
+      metadata: Optional[dict[str, Any]] = None,
   ) -> InvocationContext:
     """Sets up the context for a resumed invocation.
 
@@ -1234,6 +1253,7 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: The invocation id to resume.
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
+      metadata: Optional per-request metadata to pass to callbacks.
 
     Returns:
       The invocation context for the resumed invocation.
@@ -1259,6 +1279,7 @@ async def _setup_context_for_resumed_invocation(
         new_message=user_message,
         run_config=run_config,
         invocation_id=invocation_id,
+        metadata=metadata,
     )
     # Step 3: Maybe handle new message.
     if new_message:
@@ -1303,6 +1324,7 @@ def _new_invocation_context(
       new_message: Optional[types.Content] = None,
       live_request_queue: Optional[LiveRequestQueue] = None,
       run_config: Optional[RunConfig] = None,
+      metadata: Optional[dict[str, Any]] = None,
   ) -> InvocationContext:
     """Creates a new invocation context.
 
@@ -1312,6 +1334,7 @@ def _new_invocation_context(
         new_message: The new message for the context.
         live_request_queue: The live request queue for the context.
         run_config: The run config for the context.
+        metadata: Optional per-request metadata for the context.
 
     Returns:
         The new invocation context.
@@ -1343,6 +1366,7 @@ def _new_invocation_context(
         live_request_queue=live_request_queue,
         run_config=run_config,
         resumability_config=self.resumability_config,
+        metadata=metadata,
     )
 
   def _new_invocation_context_for_live(