In this tutorial, we explore the advanced capacities of Google agents' development kit (ADK) By building a multi-agent system equipped with specialized roles and tools. We guide you by the creation of agents adapted to tasks such as web research, mathematical calculation, data analysis and content creation. By integrating Google Search, asynchronous execution and modular architecture, we demonstrate how to orchestrate a powerful agent workflow and ready for production using the Gemini model. Our goal is to help you understand how ADK can be used to create changing and intelligent systems that are adapted to business applications. 🧵 Discover the Complete codes here
!pip install google-adk
import os
import asyncio
import json
from typing import List, Dict, Any
from dataclasses import dataclass
from google.adk.agents import Agent, LlmAgent
from google.adk.tools import google_search
def get_api_key():
"""Get API key from user input or environment variable"""
api_key = os.getenv("GOOGLE_API_KEY")
if not api_key:
from getpass import getpass
api_key = getpass("Enter your Google API Key: ")
if not api_key:
raise ValueError("API key is required to run this tutorial")
os.environ("GOOGLE_API_KEY") = api_key
return api_key
We start by installing the Google-Adk package and importing the libraries necessary to create our agent system. To authenticate our access, we collect the API Google key from the environment or invite it safely using the GetPass module. This ensures that our agents can interact with Google's tools and services transparently. 🧵 Discover the Complete codes here
@dataclass
class TaskResult:
"""Data structure for task results"""
agent_name: str
task: str
result: str
metadata: Dict(str, Any) = None
class AdvancedADKTutorial:
"""Main tutorial class demonstrating ADK capabilities"""
def __init__(self):
self.model = "gemini-1.5-flash"
self.agents = {}
self.results = ()
def create_specialized_agents(self):
"""Create a multi-agent system with specialized roles"""
self.agents('researcher') = Agent(
name="researcher",
model=self.model,
instruction="""You are a research specialist. Use Google Search to find
accurate, up-to-date information. Provide concise, factual summaries with sources.
Always cite your sources and focus on the most recent and reliable information.""",
description="Specialist in web research and information gathering",
tools=(google_search)
)
self.agents('calculator') = Agent(
name="calculator",
model=self.model,
instruction="""You are a mathematics expert. Solve calculations step-by-step.
Show your work clearly and double-check results. Handle arithmetic, algebra,
geometry, statistics, and financial calculations. Always explain your reasoning.""",
description="Expert in mathematical calculations and problem solving"
)
self.agents('analyst') = Agent(
name="analyst",
model=self.model,
instruction="""You are a data analysis expert. When given numerical data:
1. Calculate basic statistics (mean, median, min, max, range, std dev)
2. Identify patterns, trends, and outliers
3. Provide business insights and interpretations
4. Show all calculations step-by-step
5. Suggest actionable recommendations based on the data""",
description="Specialist in data analysis and statistical insights"
)
self.agents('writer') = Agent(
name="writer",
model=self.model,
instruction="""You are a professional writing assistant. Help with:
- Creating clear, engaging, and well-structured content
- Business reports and executive summaries
- Technical documentation and explanations
- Content editing and improvement
Always use professional tone and proper formatting.""",
description="Expert in content creation and document writing"
)
print("✓ Created specialized agent system:")
print(f" • Researcher: Web search and information gathering")
print(f" • Calculator: Mathematical computations and analysis")
print(f" • Analyst: Data analysis and statistical insights")
print(f" • Writer: Professional content creation")
async def run_agent_with_input(self, agent, user_input):
"""Helper method to run agent with proper error handling"""
try:
if hasattr(agent, 'generate_content'):
result = await agent.generate_content(user_input)
return result.text if hasattr(result, 'text') else str(result)
elif hasattr(agent, '__call__'):
result = await agent(user_input)
return result.text if hasattr(result, 'text') else str(result)
else:
result = str(agent) + f" processed: {user_input(:50)}..."
return result
except Exception as e:
return f"Agent execution error: {str(e)}"
async def demonstrate_single_agent_research(self):
"""Demonstrate single agent research capabilities"""
print("\n=== Single Agent Research Demo ===")
query = "What are the latest developments in quantum computing breakthroughs in 2025?"
print(f"Research Query: {query}")
try:
response_text = await self.run_agent_with_input(
agent=self.agents('researcher'),
user_input=query
)
summary = response_text(:300) + "..." if len(response_text) > 300 else response_text
task_result = TaskResult(
agent_name="researcher",
task="Quantum Computing Research",
result=summary
)
self.results.append(task_result)
print(f"✓ Research Complete: {summary}")
return response_text
except Exception as e:
error_msg = f"Research failed: {str(e)}"
print(f"❌ {error_msg}")
return error_msg
async def demonstrate_calculator_agent(self):
"""Demonstrate mathematical calculation capabilities"""
print("\n=== Calculator Agent Demo ===")
calc_problem = """Calculate the compound annual growth rate (CAGR) for an investment
that grows from $50,000 to $125,000 over 8 years. Use the formula:
CAGR = (Ending Value / Beginning Value)^(1/number of years) - 1
Express the result as a percentage."""
print("Math Problem: CAGR Calculation")
try:
response_text = await self.run_agent_with_input(
agent=self.agents('calculator'),
user_input=calc_problem
)
summary = response_text(:250) + "..." if len(response_text) > 250 else response_text
task_result = TaskResult(
agent_name="calculator",
task="CAGR Calculation",
result=summary
)
self.results.append(task_result)
print(f"✓ Calculation Complete: {summary}")
return response_text
except Exception as e:
error_msg = f"Calculation failed: {str(e)}"
print(f"❌ {error_msg}")
return error_msg
async def demonstrate_data_analysis(self):
"""Demonstrate data analysis capabilities"""
print("\n=== Data Analysis Agent Demo ===")
data_task = """Analyze this quarterly sales data for a tech startup (in thousands USD):
Q1 2025: $125K, Q2 2025: $143K, Q3 2025: $167K, Q4 2025: $152K
Q1 2025: $187K, Q2 2025: $214K, Q3 2025: $239K, Q4 2025: $263K
Calculate growth trends, identify patterns, and provide business insights."""
print("Data Analysis: Quarterly Sales Trends")
try:
response_text = await self.run_agent_with_input(
agent=self.agents('analyst'),
user_input=data_task
)
summary = response_text(:250) + "..." if len(response_text) > 250 else response_text
task_result = TaskResult(
agent_name="analyst",
task="Sales Data Analysis",
result=summary
)
self.results.append(task_result)
print(f"✓ Analysis Complete: {summary}")
return response_text
except Exception as e:
error_msg = f"Analysis failed: {str(e)}"
print(f"❌ {error_msg}")
return error_msg
async def demonstrate_content_creation(self):
"""Demonstrate content creation capabilities"""
print("\n=== Content Creation Agent Demo ===")
writing_task = """Create a brief executive summary (2-3 paragraphs) for a board presentation
that combines the key findings from:
1. Recent quantum computing developments
2. Strong financial growth trends showing 58% year-over-year growth
3. Recommendations for strategic planning
Use professional business language suitable for C-level executives."""
print("Content Creation: Executive Summary")
try:
response_text = await self.run_agent_with_input(
agent=self.agents('writer'),
user_input=writing_task
)
summary = response_text(:250) + "..." if len(response_text) > 250 else response_text
task_result = TaskResult(
agent_name="writer",
task="Executive Summary",
result=summary
)
self.results.append(task_result)
print(f"✓ Content Created: {summary}")
return response_text
except Exception as e:
error_msg = f"Content creation failed: {str(e)}"
print(f"❌ {error_msg}")
return error_msg
def display_comprehensive_summary(self):
"""Display comprehensive tutorial summary and results"""
print("\n" + "="*70)
print("🚀 ADVANCED ADK TUTORIAL - COMPREHENSIVE SUMMARY")
print("="*70)
print(f"\n📊 EXECUTION STATISTICS:")
print(f" • Total agents created: {len(self.agents)}")
print(f" • Total tasks completed: {len(self.results)}")
print(f" • Model used: {self.model} (Free Tier)")
print(f" • Runner type: Direct Agent Execution")
print(f"\n🤖 AGENT CAPABILITIES DEMONSTRATED:")
print(" • Advanced web research with Google Search integration")
print(" • Complex mathematical computations and financial analysis")
print(" • Statistical data analysis with business insights")
print(" • Professional content creation and documentation")
print(" • Asynchronous agent execution and error handling")
print(f"\n🛠️ KEY ADK FEATURES COVERED:")
print(" • Agent() class with specialized instructions")
print(" • Built-in tool integration (google_search)")
print(" • InMemoryRunner for agent execution")
print(" • Async/await patterns for concurrent operations")
print(" • Professional error handling and logging")
print(" • Modular, scalable agent architecture")
print(f"\n📋 TASK RESULTS SUMMARY:")
for i, result in enumerate(self.results, 1):
print(f" {i}. {result.agent_name.title()}: {result.task}")
print(f" Result: {result.result(:100)}...")
print(f"\n🎯 PRODUCTION READINESS:")
print(" • Code follows ADK best practices")
print(" • Ready for deployment on Cloud Run")
print(" • Compatible with Vertex AI Agent Engine")
print(" • Scalable multi-agent architecture")
print(" • Enterprise-grade error handling")
print(f"\n🔗 NEXT STEPS:")
print(" • Explore sub-agent delegation with LlmAgent")
print(" • Add custom tools and integrations")
print(" • Deploy to Google Cloud for production use")
print(" • Implement persistent memory and sessions")
print("="*70)
print("✅ Tutorial completed successfully! Happy Agent Building! 🎉")
print("="*70)
We define a Taskresult data structure to store the outputs of each agent. Then, we build a multi-agent system using Google Adk, attributing specialized roles such as the researcher, the calculator, the analyst and the writer. Thanks to asynchronous methods, we demonstrate the capacities of each agent and comply a final summary of their performance and their ideas. 🧵 Discover the Complete codes here
async def main():
"""Main tutorial execution function"""
print("🚀 Google ADK Python - Advanced Tutorial")
print("=" * 50)
try:
api_key = get_api_key()
print("✅ API key configured successfully")
except Exception as e:
print(f"❌ Error: {e}")
return
tutorial = AdvancedADKTutorial()
tutorial.create_specialized_agents()
print(f"\n🎯 Running comprehensive agent demonstrations...")
await tutorial.demonstrate_single_agent_research()
await tutorial.demonstrate_calculator_agent()
await tutorial.demonstrate_data_analysis()
await tutorial.demonstrate_content_creation()
tutorial.display_comprehensive_summary()
def run_tutorial():
"""Run the tutorial in Jupyter/Colab environment"""
import asyncio
try:
from IPython import get_ipython
if get_ipython() is not None:
return asyncio.create_task(main())
except ImportError:
pass
return asyncio.run(main())
if __name__ == "__main__":
try:
loop = asyncio.get_running_loop()
print("Detected Notebook environment. Please run: await main()")
except RuntimeError:
asyncio.run(main())
await main()
We finalize the tutorial by defining the main function (), which initializes the system, performs all agent demonstrations and displays a summary. We guarantee compatibility with script and notebook environments, which allows us to execute everything in a transparent manner with Await Main () in Colab.
In conclusion, we managed to create and deploy a fully functional multi-aging system using Google Adk. We have seen our agents manage a diversified range of tasks, carrying out real -time research and the resolution of complex financial equations to the analysis of data trends and the generation of executives summaries. We also emphasize how the ADK framework supports errors, extensibility and transparent integration with tools. Thanks to this practical experience, we gain confidence in the use of ADK to develop solutions based on robust agents for real problems, and we are delighted to explore even more advanced orchestration and deployment strategies.
Discover the Complete codes here. All the merit of this research goes to researchers in this project. Also, don't hesitate to follow us Twitter And don't forget to join our Subseubdredit 100k + ml and subscribe to Our newsletter.
🧵 You can also like Nvidia Open Cosmos Diffusionrender (Check it now)
Asif Razzaq is the CEO of Marktechpost Media Inc .. as a visionary entrepreneur and engineer, AIF undertakes to exploit the potential of artificial intelligence for social good. His most recent company is the launch of an artificial intelligence media platform, Marktechpost, which stands out from its in-depth coverage of automatic learning and in-depth learning news which are both technically solid and easily understandable by a large audience. The platform has more than 2 million monthly views, illustrating its popularity with the public.
