Google ADK Tools: Built-ins, Custom Functions, and MCP Integration

from google.adk.agents import LlmAgent
from google.adk.tools import google_search, code_execution
 
research_agent = LlmAgent(
    name='researcher',
    model='gemini-2.0-flash',
    instruction='You are a research assistant. Use search to find current information.',
    tools=[google_search],  # enable web search
)
 
data_agent = LlmAgent(
    name='data_analyst',
    model='gemini-2.0-flash',
    instruction='Analyse data and produce charts. You can execute Python code.',
    tools=[code_execution],  # enable Python sandbox
)
 

google_search requires a Google Search API key and Custom Search Engine ID. code_execution runs in a Vertex AI sandbox — it does not execute locally and incurs additional costs.

from google.adk.tools import FunctionTool
import requests
 
def get_stock_price(ticker: str) -> dict:
    """
    Returns the current stock price for a given ticker symbol.
 
    Args:
        ticker: The stock ticker symbol, e.g. 'AAPL' or 'GOOGL'.
 
    Returns:
        A dict with 'ticker', 'price', and 'currency' keys.
    """
    # Replace with a real API call
    return {'ticker': ticker, 'price': 185.23, 'currency': 'USD'}
 
stock_tool = FunctionTool(get_stock_price)
 
finance_agent = LlmAgent(
    name='finance_agent',
    model='gemini-2.0-flash',
    instruction='Help users with stock price queries.',
    tools=[stock_tool],
)
 

Write docstrings in the Args: / Returns: Google style. ADK parses them to generate descriptions for each parameter — these descriptions appear in the tool schema the LLM sees. Better descriptions = fewer tool call mistakes.

import httpx
from google.adk.tools import FunctionTool
 
async def fetch_weather(city: str) -> dict:
    """
    Fetches current weather for a city.
 
    Args:
        city: The city name, e.g. 'London' or 'New York'.
 
    Returns:
        Dict with 'temperature_c', 'condition', and 'humidity_pct'.
    """
    async with httpx.AsyncClient() as client:
        r = await client.get(f'https://wttr.in/{city}?format=j1')
        data = r.json()
        current = data['current_condition'][0]
        return {
            'temperature_c': int(current['temp_C']),
            'condition': current['weatherDesc'][0]['value'],
            'humidity_pct': int(current['humidity']),
        }
 
weather_tool = FunctionTool(fetch_weather)
 

from google.adk.tools import ToolContext
 
# Define CRM tools as a group
def get_customer(customer_id: str) -> dict:
    """Fetches a customer record by ID."""
    return {'id': customer_id, 'name': 'Alice', 'tier': 'premium'}
 
def update_customer_note(customer_id: str, note: str) -> str:
    """Adds a note to a customer's account."""
    return f'Note added to customer {customer_id}.'
 
crm_tools = [FunctionTool(get_customer), FunctionTool(update_customer_note)]
 
# Reuse across agents
support_agent = LlmAgent(
    name='support',
    model='gemini-2.0-flash',
    instruction='Handle customer support queries.',
    tools=crm_tools,  # shared toolset
)
billing_agent = LlmAgent(
    name='billing',
    model='gemini-2.0-flash',
    instruction='Handle billing questions.',
    tools=crm_tools,  # same toolset
)
 

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
 
async def create_agent_with_mcp():
    # Connect to the filesystem MCP server
    mcp_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command='npx',
            args=['-y', '@modelcontextprotocol/server-filesystem', '/tmp/workspace'],
        )
    )
 
    agent = LlmAgent(
        name='file_agent',
        model='gemini-2.0-flash',
        instruction='Help users manage files in the workspace directory.',
        tools=mcp_tools,
    )
 
    return agent, exit_stack
 
async def main():
    agent, exit_stack = await create_agent_with_mcp()
    async with exit_stack:
        from google.adk.runners import InMemoryRunner
        runner = InMemoryRunner(agent=agent, app_name='file_app')
        # ... run your agent
 

Always use the exit_stack context manager when using MCPToolset. It ensures the MCP subprocess is cleanly terminated when your agent finishes.

# Enable verbose tool call logging
import logging
logging.getLogger('google.adk').setLevel(logging.DEBUG)
 
# Or inspect events from the runner
from google.adk.events import Event
 
async for event in runner.run_async(user_id='u1', session_id='s1', new_message=msg):
    if event.get_function_calls():  # LLM requested a tool call
        print('Tool calls:', event.get_function_calls())
    if event.get_function_responses():  # Tool returned a result
        print('Tool responses:', event.get_function_responses())