fix: Copy the original function call args before passing it to callback or tools to avoid being modified

PiperOrigin-RevId: 788192459

Author: seanzhougoogle

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

@@ -152,7 +152,8 @@ async def handle_function_calls_async(
     with tracer.start_as_current_span(f'execute_tool {tool.name}'):
       # do not use "args" as the variable name, because it is a reserved keyword
       # in python debugger.
-      function_args = function_call.args or {}
+      # make a copy to avoid being modified.
+      function_args = dict(function_call.args) if function_call.args else {}
 
       # Step 1: Check if plugin before_tool_callback overrides the function
       # response.
@@ -277,7 +278,8 @@ async def handle_function_calls_live(
     with tracer.start_as_current_span(f'execute_tool {tool.name}'):
       # do not use "args" as the variable name, because it is a reserved keyword
       # in python debugger.
-      function_args = function_call.args or {}
+      # make a copy to avoid being modified.
+      function_args = dict(function_call.args) if function_call.args else {}
       function_response = None
 
       # Handle before_tool_callbacks - iterate through the canonical callback

tests/unittests/flows/llm_flows/test_functions_simple.py

@@ -392,3 +392,171 @@ def test_find_function_call_event_multiple_function_responses():
   # Should return the first matching function call event found
   result = find_matching_function_call(events)
   assert result == call_event1  # First match (func_123)
+
+
+@pytest.mark.asyncio
+async def test_function_call_args_not_modified():
+  """Test that function_call.args is not modified when making a copy."""
+  from google.adk.flows.llm_flows.functions import handle_function_calls_async
+  from google.adk.flows.llm_flows.functions import handle_function_calls_live
+
+  def simple_fn(**kwargs) -> dict:
+    return {'result': 'test'}
+
+  tool = FunctionTool(simple_fn)
+  model = testing_utils.MockModel.create(responses=[])
+  agent = Agent(
+      name='test_agent',
+      model=model,
+      tools=[tool],
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+
+  # Create original args that we want to ensure are not modified
+  original_args = {'param1': 'value1', 'param2': 42}
+  function_call = types.FunctionCall(name=tool.name, args=original_args)
+  content = types.Content(parts=[types.Part(function_call=function_call)])
+  event = Event(
+      invocation_id=invocation_context.invocation_id,
+      author=agent.name,
+      content=content,
+  )
+  tools_dict = {tool.name: tool}
+
+  # Test handle_function_calls_async
+  result_async = await handle_function_calls_async(
+      invocation_context,
+      event,
+      tools_dict,
+  )
+
+  # Verify original args are not modified
+  assert function_call.args == original_args
+  assert function_call.args is not original_args  # Should be a copy
+
+  # Test handle_function_calls_live
+  result_live = await handle_function_calls_live(
+      invocation_context,
+      event,
+      tools_dict,
+  )
+
+  # Verify original args are still not modified
+  assert function_call.args == original_args
+  assert function_call.args is not original_args  # Should be a copy
+
+  # Both should return valid results
+  assert result_async is not None
+  assert result_live is not None
+
+
+@pytest.mark.asyncio
+async def test_function_call_args_none_handling():
+  """Test that function_call.args=None is handled correctly."""
+  from google.adk.flows.llm_flows.functions import handle_function_calls_async
+  from google.adk.flows.llm_flows.functions import handle_function_calls_live
+
+  def simple_fn(**kwargs) -> dict:
+    return {'result': 'test'}
+
+  tool = FunctionTool(simple_fn)
+  model = testing_utils.MockModel.create(responses=[])
+  agent = Agent(
+      name='test_agent',
+      model=model,
+      tools=[tool],
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+
+  # Create function call with None args
+  function_call = types.FunctionCall(name=tool.name, args=None)
+  content = types.Content(parts=[types.Part(function_call=function_call)])
+  event = Event(
+      invocation_id=invocation_context.invocation_id,
+      author=agent.name,
+      content=content,
+  )
+  tools_dict = {tool.name: tool}
+
+  # Test handle_function_calls_async
+  result_async = await handle_function_calls_async(
+      invocation_context,
+      event,
+      tools_dict,
+  )
+
+  # Test handle_function_calls_live
+  result_live = await handle_function_calls_live(
+      invocation_context,
+      event,
+      tools_dict,
+  )
+
+  # Both should return valid results even with None args
+  assert result_async is not None
+  assert result_live is not None
+
+
+@pytest.mark.asyncio
+async def test_function_call_args_copy_behavior():
+  """Test that modifying the copied args doesn't affect the original."""
+  from google.adk.flows.llm_flows.functions import handle_function_calls_async
+  from google.adk.flows.llm_flows.functions import handle_function_calls_live
+
+  def simple_fn(test_param: str, other_param: int) -> dict:
+    # Modify the args to test that the copy prevents affecting the original
+    return {
+        'result': 'test',
+        'received_args': {'test_param': test_param, 'other_param': other_param},
+    }
+
+  tool = FunctionTool(simple_fn)
+  model = testing_utils.MockModel.create(responses=[])
+  agent = Agent(
+      name='test_agent',
+      model=model,
+      tools=[tool],
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+
+  # Create original args
+  original_args = {'test_param': 'original_value', 'other_param': 123}
+  function_call = types.FunctionCall(name=tool.name, args=original_args)
+  content = types.Content(parts=[types.Part(function_call=function_call)])
+  event = Event(
+      invocation_id=invocation_context.invocation_id,
+      author=agent.name,
+      content=content,
+  )
+  tools_dict = {tool.name: tool}
+
+  # Test handle_function_calls_async
+  result_async = await handle_function_calls_async(
+      invocation_context,
+      event,
+      tools_dict,
+  )
+
+  # Verify original args are unchanged
+  assert function_call.args == original_args
+  assert function_call.args['test_param'] == 'original_value'
+
+  # Verify the tool received the args correctly
+  assert result_async is not None
+  response = result_async.content.parts[0].function_response.response
+
+  # Check if the response has the expected structure
+  assert 'received_args' in response
+  received_args = response['received_args']
+  assert 'test_param' in received_args
+  assert received_args['test_param'] == 'original_value'
+  assert received_args['other_param'] == 123
+  assert (
+      function_call.args['test_param'] == 'original_value'
+  )  # Original unchanged