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

src/google/adk/agents/llm_agent.py

@@ -352,6 +352,27 @@ class LlmAgent(BaseAgent, abc.ABC):
     instruction and input
   """
 
+  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
   input_schema: Optional[type[BaseModel]] = None
   """The input schema when agent is used as a tool."""
@@ -957,6 +978,17 @@ def __maybe_save_output_to_state(self, event: Event):
   def __model_validator_after(self) -> LlmAgent:
     return self
 
+  @field_validator('include_sources', mode='after')
+  @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/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,6 +79,7 @@ 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)
@@ -89,6 +91,7 @@ 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,
       )
 
     # Add instruction-related contents to proper position in conversation
@@ -504,6 +507,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 +614,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 +628,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,6 +717,7 @@ 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,
 ) -> list[types.Content]:
   """Get contents for the current turn only (no conversation history).
 
@@ -712,6 +753,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 []

tests/unittests/agents/test_llm_agent_include_contents.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Unit tests for LlmAgent include_contents field behavior."""
+"""Unit tests for LlmAgent include_contents and include_sources field behavior."""
 
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.agents.sequential_agent import SequentialAgent
@@ -241,3 +241,131 @@ async def test_include_contents_none_sequential_agents():
   assert any(
       "Agent1 response" in str(content) for _, content in agent2_contents
   )
+
+
+# ---------------------------------------------------------------------------
+# include_sources: field validation
+# ---------------------------------------------------------------------------
+
+
+def test_include_sources_empty_list_raises():
+  """include_sources=[] must raise ValueError — use None to disable filtering."""
+  with pytest.raises(ValueError, match='include_sources=\\[\\]'):
+    LlmAgent(
+        name='agent',
+        model='gemini-2.5-flash',
+        include_sources=[],
+    )
+
+
+def test_include_sources_none_is_accepted():
+  """include_sources=None (default) must not raise."""
+  agent = LlmAgent(
+      name='agent', model='gemini-2.5-flash', include_sources=None
+  )
+  assert agent.include_sources is None
+
+
+# ---------------------------------------------------------------------------
+# include_sources: integration — user-only in sequential pipeline
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_include_sources_user_only_drops_upstream_agent_entries():
+  """Downstream agent with include_sources=['user'] receives only the human user message."""
+  agent1_model = testing_utils.MockModel.create(
+      responses=['Upstream agent reply']
+  )
+  agent1 = LlmAgent(
+      name='upstream',
+      model=agent1_model,
+      instruction='You are upstream',
+  )
+
+  agent2_model = testing_utils.MockModel.create(
+      responses=['Downstream response']
+  )
+  agent2 = LlmAgent(
+      name='downstream',
+      model=agent2_model,
+      include_sources=['user'],
+      instruction='You are downstream',
+  )
+
+  sequential = SequentialAgent(
+      name='pipeline', sub_agents=[agent1, agent2]
+  )
+  runner = testing_utils.InMemoryRunner(sequential)
+  runner.run('Original user request')
+
+  agent2_contents = testing_utils.simplify_contents(
+      agent2_model.requests[0].contents
+  )
+
+  # User message must be present
+  assert any(
+      'Original user request' in str(c) for _, c in agent2_contents
+  )
+  # Upstream agent's narrative entry must be absent
+  assert not any(
+      'Upstream agent reply' in str(c) for _, c in agent2_contents
+  )
+  assert not any('For context:' in str(c) for _, c in agent2_contents)
+
+
+# ---------------------------------------------------------------------------
+# include_sources: composing with include_contents='default' — multi-turn
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_include_sources_user_self_drops_upstream_across_turns():
+  """include_sources=['user','self'] + include_contents='default' (full history):
+  downstream agent sees all user messages and its own prior turns, but no
+  narrative entries from the upstream agent across multiple invocations.
+  """
+  agent1_model = testing_utils.MockModel.create(
+      responses=['Turn1 upstream reply', 'Turn2 upstream reply']
+  )
+  agent1 = LlmAgent(
+      name='upstream',
+      model=agent1_model,
+      instruction='You are upstream',
+  )
+
+  agent2_model = testing_utils.MockModel.create(
+      responses=['Turn1 downstream', 'Turn2 downstream']
+  )
+  agent2 = LlmAgent(
+      name='downstream',
+      model=agent2_model,
+      include_sources=['user', 'self'],
+      instruction='You are downstream',
+  )
+
+  sequential = SequentialAgent(
+      name='pipeline', sub_agents=[agent1, agent2]
+  )
+  runner = testing_utils.InMemoryRunner(sequential)
+  runner.run('Turn 1 user message')
+  runner.run('Turn 2 user message')
+
+  # Second invocation of downstream agent — should see user messages + own
+  # prior turn, but not upstream's narrative entries.
+  agent2_second_contents = testing_utils.simplify_contents(
+      agent2_model.requests[1].contents
+  )
+
+  # User messages must be present
+  assert any(
+      'Turn 1 user message' in str(c) for _, c in agent2_second_contents
+  )
+  assert any(
+      'Turn 2 user message' in str(c) for _, c in agent2_second_contents
+  )
+  # Upstream agent's narrative entries must be absent
+  assert not any(
+      'upstream reply' in str(c).lower() for _, c in agent2_second_contents
+  )
+  assert not any('For context:' in str(c) for _, c in agent2_second_contents)