feat(agents): add include_sources for per-agent content source filtering

src/google/adk/agents/llm_agent.py

@@ -343,13 +343,36 @@ class LlmAgent(BaseAgent, abc.ABC):
   """Disallows LLM-controlled transferring to the peer agents."""
   # LLM-based agent transfer configs - End
 
-  include_contents: Literal['default', 'none'] = 'default'
+  include_contents: Literal['default', 'current', 'none'] = 'default'
   """Controls content inclusion in model requests.
 
   Options:
-    default: Model receives relevant conversation history
-    none: Model receives no prior history, operates solely on current
-    instruction and input
+    default: Model receives full conversation history.
+    current: Model receives all events since the last user message,
+      including outputs from agents that ran earlier in this pipeline.
+    none: Model receives only the most recent agent or user input, with no
+      prior conversation history.
+  """
+
+  include_sources: Optional[list[str]] = None
+  """Allowlist of content sources to include in model requests.
+
+  Orthogonal to include_contents (temporal window); this controls which
+  sources are kept from within that window.
+
+  Options:
+    None (default): all sources pass through — backward-compatible.
+    list[str]: only content from the listed sources is kept.
+
+  Reserved source names:
+    'user'  — plain human user messages (not tool outputs)
+    'self'  — this agent's own prior model outputs
+    <name>  — any other string is matched against event.author (agent name)
+
+  Example — keep full history but only user + this agent's turns:
+    include_contents='default', include_sources=['user', 'self']
+
+  Raises ValueError if set to [] (use None to disable filtering).
   """
 
   # Controlled input/output configurations - Start
@@ -955,8 +978,29 @@ def __maybe_save_output_to_state(self, event: Event):
 
   @model_validator(mode='after')
   def __model_validator_after(self) -> LlmAgent:
+    if self.include_contents == 'none' and self.include_sources is not None:
+      warnings.warn(
+          "include_contents='none' with include_sources may produce empty"
+          ' context: the turn boundary is the last user OR other-agent event,'
+          ' and if that event is filtered by include_sources the context will'
+          " be empty. Use include_contents='current' to anchor at the last"
+          ' user message instead.',
+          UserWarning,
+          stacklevel=2,
+      )
     return self
 
+  @field_validator('include_sources', mode='after')  # type: ignore[misc]
+  @classmethod
+  def _validate_include_sources(
+      cls, v: Optional[list[str]]
+  ) -> Optional[list[str]]:
+    if v is not None and len(v) == 0:
+      raise ValueError(
+          'include_sources=[] keeps nothing. Use None to disable filtering.'
+      )
+    return v
+
   @field_validator('generate_content_config', mode='after')
   @classmethod
   def validate_generate_content_config(

src/google/adk/agents/llm_agent_config.py

@@ -121,7 +121,7 @@ def _validate_model_sources(self) -> LlmAgentConfig:
       default=None, description='Optional. LlmAgent.output_key.'
   )
 
-  include_contents: Literal['default', 'none'] = Field(
+  include_contents: Literal['default', 'current', 'none'] = Field(
       default='default', description='Optional. LlmAgent.include_contents.'
   )
 

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

@@ -68,6 +68,7 @@ async def run_async(
     instruction_related_contents = llm_request.contents
 
     is_single_turn = getattr(agent, 'mode', None) == 'single_turn'
+    source_filter = getattr(agent, 'include_sources', None)
     if agent.include_contents == 'default':
       # Include full conversation history
       llm_request.contents = _get_contents(
@@ -78,9 +79,13 @@ async def run_async(
           isolation_scope=invocation_context.isolation_scope,
           is_single_turn=is_single_turn,
           user_content=invocation_context.user_content,
+          source_filter=source_filter,
       )
     else:
-      # Include current turn context only (no conversation history)
+      # 'current': anchor at last user message — all sibling agent outputs
+      #            within this invocation are included.
+      # 'none':    anchor at last turn boundary (user OR other-agent event).
+      stop_at_user_only = agent.include_contents == 'current'
       llm_request.contents = _get_current_turn_contents(
           invocation_context.branch,
           invocation_context.session.events,
@@ -89,6 +94,8 @@ async def run_async(
           isolation_scope=invocation_context.isolation_scope,
           is_single_turn=is_single_turn,
           user_content=invocation_context.user_content,
+          source_filter=source_filter,
+          stop_at_user_only=stop_at_user_only,
       )
 
     # Add instruction-related contents to proper position in conversation
@@ -504,6 +511,7 @@ def _get_contents(
     isolation_scope: Optional[str] = None,
     is_single_turn: bool = False,
     user_content: Optional[types.Content] = None,
+    source_filter: Optional[list[str]] = None,
 ) -> list[types.Content]:
   """Get the contents for the LLM request.
 
@@ -610,6 +618,7 @@ def _get_contents(
         accumulated_output_transcription = ''
 
     is_other_reply = _is_other_agent_reply(agent_name, event)
+    other_fc_author = None  # set when is_other_reply via FC attribution
 
     # Check if it's a FunctionResponse for another agent
     if not is_other_reply and event.content:
@@ -623,8 +632,43 @@ def _get_contents(
               and call_author != 'user'
           ):
             is_other_reply = True
+            other_fc_author = call_author
             break
 
+    if source_filter is not None:
+      if is_other_reply:
+        if event.author != 'user':
+          # In live mode the current agent's own events are also classified as
+          # other_reply (see _is_other_agent_reply). Map the actual agent name
+          # to the 'self' reserved name so source_filter=['self'] works.
+          effective_source = (
+              'self' if event.author == agent_name else event.author
+          )
+          if effective_source not in source_filter:
+            continue
+        else:
+          # 'user'-authored FC response to another agent's call.
+          # other_fc_author was resolved above — no second iteration needed.
+          # _present_other_agent_message converts it to text, so no raw
+          # function_response survives — but drop it when its call author is
+          # filtered to avoid "[agent_b] returned X" with no visible preceding
+          # "[agent_b] called tool Y".
+          if other_fc_author and other_fc_author not in source_filter:
+            continue
+      elif event.content:
+        if event.content.role == 'model':
+          if 'self' not in source_filter:
+            continue
+        elif event.content.role == 'user':
+          if _content_contains_function_response(event.content):
+            # FC responses are paired with the current agent's own tool calls
+            # (role='model'). Tie them to 'self' so dropping 'self' drops both
+            # sides of the pair and avoids orphaned function_response parts.
+            if 'self' not in source_filter:
+              continue
+          elif 'user' not in source_filter:
+            continue
+
     if is_other_reply:
       if converted_event := _present_other_agent_message(event):
         filtered_events.append(converted_event)
@@ -677,33 +721,42 @@ def _get_current_turn_contents(
     is_single_turn: bool = False,
     isolation_scope: Optional[str] = None,
     user_content: Optional[types.Content] = None,
+    source_filter: Optional[list[str]] = None,
+    stop_at_user_only: bool = False,
 ) -> list[types.Content]:
   """Get contents for the current turn only (no conversation history).
 
-  When include_contents='none', we want to include:
-  - The current user input
-  - Tool calls and responses from the current turn
-  But exclude conversation history from previous turns.
-
-  In multi-agent scenarios, the "current turn" for an agent starts from an
-  actual user or from another agent.
+  Used by include_contents='none' and 'current'. Both exclude prior-session
+  history; they differ in the turn boundary:
+    'none'    (stop_at_user_only=False): last user OR other-agent event.
+    'current' (stop_at_user_only=True):  last user event only.
 
   Args:
     current_branch: The current branch of the agent.
     events: A list of all session events.
     agent_name: The name of the agent.
     preserve_function_call_ids: Whether to preserve function call ids.
+    stop_at_user_only: When True, anchor only at user events ('current' mode).
 
   Returns:
-    A list of contents for the current turn only, preserving context needed
-    for proper tool execution while excluding conversation history.
+    A list of contents from the turn boundary forward. Returns [] if no
+    qualifying boundary event is found.
   """
-  # Find the latest event that starts the current turn and process from there
+  # Find the latest event that starts the current turn and process from there.
+  # stop_at_user_only=True ('current' mode): anchor at last user message,
+  # so all sibling agent outputs within this invocation are included.
+  # stop_at_user_only=False ('none' mode): anchor at last user OR other-agent.
   for i in range(len(events) - 1, -1, -1):
     event = events[i]
-    if _should_include_event_in_context(
-        current_branch, event, isolation_scope=isolation_scope
-    ) and (event.author == 'user' or _is_other_agent_reply(agent_name, event)):
+    is_turn_start = event.author == 'user' or (
+        not stop_at_user_only and _is_other_agent_reply(agent_name, event)
+    )
+    if (
+        _should_include_event_in_context(
+            current_branch, event, isolation_scope=isolation_scope
+        )
+        and is_turn_start
+    ):
       return _get_contents(
           current_branch,
           events[i:],
@@ -712,6 +765,7 @@ def _get_current_turn_contents(
           isolation_scope=isolation_scope,
           is_single_turn=is_single_turn,
           user_content=user_content,
+          source_filter=source_filter,
       )
 
   return []