How to execute tools programmatically

[lucasbarzotto-axonify]

Hi, could you please provide an example on how to execute tools (e.g. MCP tools) in parallel programmatically inside a custom agent, keeping tracing / events working the same as if it was executed by an LLM agent?

Could we utilize the handle_function_call_list_async method from src/google/adk/flows/llm_flows/functions.py for that?

Thanks!

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: Yes, you are on the right track. The handle_function_call_list_async function is the correct way to programmatically execute a list of tools in parallel while ensuring that tracing and event logging work as expected.

---

Hi! That's a great question.

You are correct! To execute tools programmatically in parallel from within a custom agent, you can and should use the handle_function_call_list_async method. [1] This function is designed to handle the asynchronous execution of multiple tool calls, and because it's part of the ADK's internal flow management, it will automatically integrate with the existing tracing and event-handling system.

Here’s a conceptual guide on how you could use it inside a custom agent:

1. Create a Custom Agent: If you haven't already, you'll need a custom agent that inherits from BaseAgent. You will be implementing your logic in the _run_async_impl method. [2]

2. Construct Function Calls: Programmatically create a list of google.genai.types.FunctionCall objects. Each object needs the name of the tool you want to call and the args (as a dictionary) for that tool.

   from google.generativeai.types import FunctionCall
   
   function_calls = [
       FunctionCall(name='my_tool_1', args={'param_a': 'value_a'}),
       FunctionCall(name='my_tool_2', args={'param_b': 123}),
   ]

3. Get Necessary Context: Inside your agent's _run_async_impl method, you will have access to the invocation_context. You can get the dictionary of available tools from the agent itself.

4. Call handle_function_call_list_async: Call the function with the FunctionCall list and the context you gathered.

   from google.adk.flows.llm_flows import functions as adk_functions
   
   # Inside your custom agent's _run_async_impl(self, ...) method:
   tools_dict = self.get_tools() # Or however you access your toolset
   
   # This assumes 'invocation_context' is passed into your method
   tool_results = await adk_functions.handle_function_call_list_async(
       function_call_list=function_calls,
       tools_dict=tools_dict,
       invocation_context=invocation_context
   )

As long as the tools you are calling are defined as asynchronous (async def), this function will execute them in parallel. [3] The invocation_context ensures that all the necessary telemetry (tracing and events) is captured correctly, just as it would be if an LLM had initiated the tool calls. [4]

This approach gives you the power to create complex, programmatic workflows within the ADK framework while still benefiting from its built-in observability features.

---

[1] https://github.com/google/adk-python/blob/main/src/google/adk/flows/llm_flows/functions.py
[2] https://google.github.io/adk-docs/agents/custom-agents/
[3] https://google.github.io/adk-docs/tools/performance/
[4] https://github.com/google/adk-python/blob/main/tests/unittests/flows/llm_flows/test_tool_telemetry.py