diff --git a/src/google/adk/agents/run_config.py b/src/google/adk/agents/run_config.py
index 9fe82fabf1..754970537e 100644
--- a/src/google/adk/agents/run_config.py
+++ b/src/google/adk/agents/run_config.py
@@ -94,6 +94,11 @@ class RunConfig(BaseModel):
   session_resumption: Optional[types.SessionResumptionConfig] = None
   """Configures session resumption mechanism. Only support transparent session resumption mode now."""
 
+  context_window_compression: Optional[types.ContextWindowCompressionConfig] = (
+      None
+  )
+  """Configuration for context window compression. If set, this will enable context window compression for LLM input."""
+
   save_live_audio: bool = False
   """Saves live video and audio data to session and artifact service.
 
diff --git a/src/google/adk/flows/llm_flows/basic.py b/src/google/adk/flows/llm_flows/basic.py
index 013c7ad054..24cc7fd67c 100644
--- a/src/google/adk/flows/llm_flows/basic.py
+++ b/src/google/adk/flows/llm_flows/basic.py
@@ -79,6 +79,9 @@ async def run_async(
     llm_request.live_connect_config.session_resumption = (
         invocation_context.run_config.session_resumption
     )
+    llm_request.live_connect_config.context_window_compression = (
+        invocation_context.run_config.context_window_compression
+    )
 
     # TODO: handle tool append here, instead of in BaseTool.process_llm_request.
 
diff --git a/tests/unittests/streaming/test_live_streaming_configs.py b/tests/unittests/streaming/test_live_streaming_configs.py
index c5b3606776..ecb253e09f 100644
--- a/tests/unittests/streaming/test_live_streaming_configs.py
+++ b/tests/unittests/streaming/test_live_streaming_configs.py
@@ -586,3 +586,59 @@ def test_streaming_with_session_resumption_config():
       llm_request_sent_to_mock.live_connect_config.session_resumption.transparent
       is True
   )
+
+
+def test_streaming_with_context_window_compression_config():
+  """Test streaming with context window compression config."""
+  response = LlmResponse(turn_complete=True)
+
+  mock_model = testing_utils.MockModel.create([response])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with context window compression
+  run_config = RunConfig(
+      context_window_compression=types.ContextWindowCompressionConfig(
+          trigger_tokens=1000,
+          sliding_window=types.SlidingWindow(target_tokens=500),
+      )
+  )
+
+  live_request_queue = LiveRequestQueue()
+  live_request_queue.send_realtime(
+      blob=types.Blob(data=b'\x00\xFF', mime_type='audio/pcm')
+  )
+
+  res_events = runner.run_live(live_request_queue, run_config)
+
+  assert res_events is not None, 'Expected a list of events, got None.'
+  assert (
+      len(res_events) > 0
+  ), 'Expected at least one response, but got an empty list.'
+  assert len(mock_model.requests) == 1
+
+  # Get the request that was captured
+  llm_request_sent_to_mock = mock_model.requests[0]
+
+  # Assert that the request contained the correct configuration
+  assert llm_request_sent_to_mock.live_connect_config is not None
+  assert (
+      llm_request_sent_to_mock.live_connect_config.context_window_compression
+      is not None
+  )
+  assert (
+      llm_request_sent_to_mock.live_connect_config.context_window_compression.trigger_tokens
+      == 1000
+  )
+  assert (
+      llm_request_sent_to_mock.live_connect_config.context_window_compression.sliding_window.target_tokens
+      == 500
+  )