### Python - Google Search Agent Setup and Execution

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 5_ Tool Use (using Google Search).ipynb

This Python snippet initializes an ADK Agent with the Google Search tool and demonstrates how to run it asynchronously to answer a user's query. It requires the google-adk library and asyncio. The agent takes a user query as input and returns the search results as a text response.

```python
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

# Define variables required for Session setup and Agent execution
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"


# Define Agent with access to search tool
root_agent = ADKAgent(
   name="basic_search_agent",
   model="gemini-2.0-flash-exp",
   description="Agent to answer questions using Google Search.",
   instruction="I can answer your questions by searching the internet. Just ask me anything!",
   tools=[google_search] # Google Search is a pre-built tool to perform Google searches.
)

# Agent Interaction
async def call_agent(query):
   """
   Helper function to call the agent with a query.
   """

   # Session and Runner
   session_service = InMemorySessionService()
   session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
   runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)

   content = types.Content(role='user', parts=[types.Part(text=query)])
   events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)


   for event in events:
       if event.is_final_response():
           final_response = event.content.parts[0].text
           print("Agent Response: ", final_response)

nest_asyncio.apply()

asyncio.run(call_agent("what's the latest ai news?"))
```

--------------------------------

### Setup Session Variables for Google Search Agent (Python)

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This Python code defines essential variables for setting up a session and executing an agent, specifically for an application named 'Google Search_agent'. It includes user and session identifiers.

```python
# Define variables required for Session setup and Agent execution
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"


```

--------------------------------

### Install Python Dependencies for Agentic Design Patterns

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This command installs the required Python packages for working with agentic design patterns. It includes libraries for environment variable management, Google Generative AI integration with Langchain, and the CrewAI framework. Ensure you have pip installed and a stable internet connection for the download and installation process.

```shell
!pip install -q -U "dotenv==0.9.9" "langchain-google-genai==2.1.8" "crewai==0.150.0" "google-adk==1.8.0"
```

--------------------------------

### Configure Basic Logging

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Sets up basic logging for the application to provide information about execution flow and potential errors. The logger is configured to display timestamps, log level, and the message content.

```python
# Basic logging setup helps in debugging and tracking to execution.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
```

--------------------------------

### RAG Setup with LangChain, LangGraph, and Weaviate in Python

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Initializes a Retrieval-Augmented Generation (RAG) pipeline using LangChain and LangGraph. It includes loading documents, splitting them into chunks, setting up embeddings, and configuring a Weaviate vector store.

```python
import requests
from langchain_community.document_loaders import TextLoader
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Weaviate
from langchain_openai import ChatOpenAI
from langchain.text_splitter import CharacterTextSplitter
from langgraph.graph import StateGraph, END
import weaviate
from weaviate.embedded import EmbeddedOptions
from typing import List, TypedDict
import dotenv

dotenv.load_dotenv()

# Load and prepare data
url = "https://github.com/langchain-ai/langchain/blob/master/docs/docs/how_to/state_of_the_union.txt"
res = requests.get(url)

with open("state_of_the_union.txt", "w") as f:
    f.write(res.text)

loader = TextLoader('./state_of_the_union.txt')
documents = loader.load()

# Chunk documents
text_splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)
```

--------------------------------

### Initialize Google Generative AI Language Model

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Initializes the `ChatGoogleGenerativeAI` model, specifically the 'gemini-2.0-flash' version, with a temperature of 0 for deterministic output. Includes error handling for initialization failures.

```python
try:
   # A model with function/tool calling capabilities is required.
   llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
   print(f"✅ Language model initialized: {llm.model}")
except Exception as e:
   print(f"🛑 Error initializing language model: {e}")
   llm = None
```

--------------------------------

### Async Agent Interaction with Code Execution (Python)

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Demonstrates asynchronous interaction with an agent, including session setup, running queries, and processing events containing executable code, code execution results, and text. It handles potential errors during the agent run and adapts to different execution environments like Jupyter notebooks.

```python
import asyncio
import nest_asyncio
from google.generativeai.types import Content, Part
from jupyter_client import InMemorySessionService
from runner import Runner

# Assume APP_NAME, USER_ID, SESSION_ID, code_agent are defined elsewhere
APP_NAME = "my_app"
USER_ID = "user123"
SESSION_ID = "session456"
# Placeholder for the actual agent object
class MockAgent:
    async def generate_content(self, prompt):
        # Mock response based on prompt for demonstration
        if "(5 + 7) * 3" in prompt:
            return types.GenerateContentResponse(
                candidates=[
                    types.Candidate(
                        content=types.Content(
                            parts=[
                                types.Part(executable_code=types.ExecutableCode(code='print((5 + 7) * 3)')),
                                types.Part(code_execution_result=types.CodeExecutionResult(outcome='OUTCOME_OK', output='36')),
                                types.Part(text='36')
                            ])
                        )
                    )
                ]
            )
        elif "10 factorial" in prompt:
            return types.GenerateContentResponse(
                candidates=[
                    types.Candidate(
                        content=types.Content(
                            parts=[
                                types.Part(executable_code=types.ExecutableCode(code='import math\nprint(math.factorial(10))')),
                                types.Part(code_execution_result=types.CodeExecutionResult(outcome='OUTCOME_OK', output='3628800')),
                                types.Part(text='3628800')
                            ])
                        )
                    )
                ]
            )
        return types.GenerateContentResponse()

code_agent = MockAgent()

async def call_agent_async(query):
   # Session and Runner
   session_service = InMemorySessionService()
   session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
   runner = Runner(agent=code_agent, app_name=APP_NAME, session_service=session_service)

   content = types.Content(role='user', parts=[types.Part(text=query)])
   print(f"\n--- Running Query: {query} ---")
   final_response_text = "No final text response captured."
   try:
       # Use run_async
       async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
           print(f"Event ID: {event.id}, Author: {event.author}")

           # --- Check for specific parts FIRST ---
           has_specific_part = False
           if event.content and event.content.parts and event.is_final_response():
               for part in event.content.parts: # Iterate through all parts
                   if part.executable_code:
                       # Access the actual code string via .code
                       print(f"  Debug: Agent generated code:\n```python\n{part.executable_code.code}\n```")
                       has_specific_part = True
                   elif part.code_execution_result:
                       # Access outcome and output correctly
                       print(f"  Debug: Code Execution Result: {part.code_execution_result.outcome} - Output:\n{part.code_execution_result.output}")
                       has_specific_part = True
                   # Also print any text parts found in any event for debugging
                   elif part.text and not part.text.isspace():
                       print(f"  Text: '{part.text.strip()}'")
                       # Do not set has_specific_part=True here, as we want the final response logic below

               # --- Check for final response AFTER specific parts ---
               text_parts = [part.text for part in event.content.parts if part.text]
               final_result = "".join(text_parts)
               print(f"==> Final Agent Response: {final_result}")

   except Exception as e:
       print(f"ERROR during agent run: {e}")
   print("-" * 30)

# Main async function to run the examples
async def main():
   await call_agent_async("Calculate the value of (5 + 7) * 3")
   await call_agent_async("What is 10 factorial?")


# Execute the main async function
try:
   nest_asyncio.apply()
   asyncio.run(main())
except RuntimeError as e:
   # Handle specific error when running asyncio.run in an already running loop (like Jupyter/Colab)
   if "cannot be called from a running event loop" in str(e):
       print("\nRunning in an existing event loop (like Colab/Jupyter).")
       print("Please run `await main()` in a notebook cell instead.")
       # If in an interactive environment like a notebook, you might need to run:
       # await main()
   else:
       raise e # Re-raise other runtime errors

```

--------------------------------

### Import Core Libraries for Agentic Design Patterns

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Imports essential Python libraries for building agents, including asyncio, Langchain components, CrewAI, and Google ADK functionalities. It also handles environment variable loading and logging setup.

```python
import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging


from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

from crewai import Agent as CrewAgent, Task, Crew
from crewai.tools import tool as crew_tool

from google.adk.agents import Agent as ADKAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types
```

--------------------------------

### Define a Search Information Tool for Langchain

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Defines a Langchain tool named `search_information` that simulates searching for factual information. It takes a query string and returns a predefined string response based on the query, demonstrating tool creation and usage.

```python
# --- Define a Tool ---
@langchain_tool
def search_information(query: str) -> str:
   """
   Provides factual information on a given topic. Use this tool to find answers to phrases
   like 'capital of France' or 'weather in London?'.
   """
   print(f"\n--- 🛠️ Tool Called: search_information with query: '{query}' ---")
   # Simulate a search tool with a dictionary of predefined results.
   simulated_results = {
       "weather in london": "The weather in London is currently cloudy with a temperature of 15°C.",
       "capital of france": "The capital of France is Paris.",
       "population of earth": "The estimated population of Earth is around 8 billion people.",
       "tallest mountain": "Mount Everest is the tallest mountain above sea level.",
       "default": f"Simulated search result for '{query}': No specific information found, but the topic seems interesting."
   }
   result = simulated_results.get(query.lower(), simulated_results["default"])
   print(f"--- TOOL RESULT: {result} ---")
   return result


tools = [search_information]
```

--------------------------------

### Install Dependencies for LangChain and OpenAI

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 11_ Goal Setting and Monitoring (Goal_Setting_Iteration).ipynb

Installs necessary Python packages including langchain_openai, openai, and python-dotenv for managing API keys and LangChain functionalities. This is typically the first step in setting up the project environment.

```python
!pip install langchain_openai openai python-dotenv
```

--------------------------------

### Environment Setup and LLM Initialization

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 11_ Goal Setting and Monitoring (Goal_Setting_Iteration).ipynb

Loads environment variables, specifically the OpenAI API key, using dotenv. It then initializes the ChatOpenAI model, specifying 'gpt-4o' as the model and setting a low temperature for more deterministic output. Error handling is included for missing API keys.

```python
# MIT License
# Copyright (c) 2025 Mahtab Syed
# https://www.linkedin.com/in/mahtabsyed/
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so.

"""
Hands-On Code Example - Iteration 2
- To illustrate the Goal Setting and Monitoring pattern, we have an example using LangChain and OpenAI APIs:

Objective: Build an AI Agent which can write code for a specified use case based on specified goals:
- Accepts a coding problem (use case) in code or can be as input.
- Accepts a list of goals (e.g., "simple", "tested", "handles edge cases")  in code or can be input.
- Uses an LLM (like GPT-4o) to generate and refine Python code until the goals are met. (I am using max 5 iterations, this could be based on a set goal as well)
- To check if we have met our goals I am asking the LLM to judge this and answer just True or False which makes it easier to stop the iterations.
- Saves the final code in a .py file with a clean filename and a header comment.
"""


import os
import random
import re
from pathlib import Path
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv, find_dotenv

# 🔐 Load environment variables
_ = load_dotenv(find_dotenv())
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise EnvironmentError("❌ Please set the OPENAI_API_KEY environment variable.")

# ✅ Initialize OpenAI model
print("📡 Initializing OpenAI LLM (gpt-4o)...")
llm = ChatOpenAI(
    model="gpt-4o", # If you dont have access to got-4o use other OpenAI LLMs
    temperature=0.3,
    openai_api_key=OPENAI_API_KEY,
)

```

--------------------------------

### Create Tool-Calling Agent with Prompt and Executor

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This code sets up a tool-calling agent by defining a chat prompt template that includes placeholders for input and agent scratchpad. It then creates the agent by binding the Language Model (LLM), tools, and the prompt, and finally initializes an AgentExecutor for runtime execution.

```python
if llm:
   # This prompt template requires an `agent_scratchpad` placeholder for the agent's internal steps.
   agent_prompt = ChatPromptTemplate.from_messages([
       ("system", "You are a helpful assistant."),
       ("human", "{input}"),
       ("placeholder", "{agent_scratchpad}"),
   ])


   # Create the agent, binding the LLM, tools, and prompt together.
   agent = create_tool_calling_agent(llm, tools, agent_prompt)


   # AgentExecutor is the runtime that invokes the agent and executes the chosen tools.
   # The 'tools' argument is not needed here as they are already bound to the agent.
   agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)
```

--------------------------------

### Securely Set API Keys

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Prompts the user to securely enter their Google API key and OpenAI API key using `getpass` and sets them as environment variables. This is crucial for authenticating with the respective AI services.

```python
# UNCOMMENT
# Prompt the user securely and set API keys as an environment variables
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API key: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
```

--------------------------------

### Agent-to-Agent Communication Setup with Google ADK (Python)

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

This snippet illustrates setting up inter-agent communication using the Google ADK framework. It defines an asynchronous function to create a calendar agent capable of managing user calendars and interacting with Google Calendar API. It also defines AgentCard and AgentSkill for the Agent-to-Agent protocol, specifying the agent's capabilities and skills. Dependencies include google.adk libraries.

```python
from google.adk.agents import LlmAgent
from google.adk.tools.google_api_tool import CalendarToolset
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.memory import InMemoryMemoryService
from google.adk.artifacts import InMemoryArtifactService
import datetime
import os

async def create_agent(client_id, client_secret) -> LlmAgent:
    """Constructs the ADK agent."""
    toolset = CalendarToolset(client_id=client_id, client_secret=client_secret)
    return LlmAgent(
        model='gemini-2.0-flash-001',
        name='calendar_agent',
        description="An agent that can help manage a user's calendar",
        instruction=f"""
You are an agent that can help manage a user's calendar.

Users will request information about the state of their calendar or to make changes to
their calendar. Use the provided tools for interacting with the calendar API.

If not specified, assume the calendar the user wants is the 'primary' calendar.

When using the Calendar API tools, use well-formed RFC3339 timestamps.

Today is {datetime.datetime.now()}.
""",
        tools=await toolset.get_tools(),
    )

# Agent Card and Skill definition for A2A protocol
from google.adk.a2a import AgentCard, AgentSkill, AgentCapabilities

skill = AgentSkill(
    id='check_availability',
    name='Check Availability',
    description="Checks a user's availability for a time using their Google Calendar",
    tags=['calendar'],
    examples=['Am I free from 10am to 11am tomorrow?'],
)

agent_card = AgentCard(
    name='Calendar Agent',
    description="An agent that can manage a user's calendar",
    url=f'http://localhost:8000/',
    version='1.0.0',
    defaultInputModes=['text'],
    defaultOutputModes=['text'],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)
```

--------------------------------

### Planning and Writing with CrewAI in Python

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Demonstrates how to use CrewAI to define agents, tasks, and a crew to plan and write a summary on a given topic. It utilizes Langchain's ChatOpenAI for the language model and dotenv for environment variables.

```python
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# Define the language model
llm = ChatOpenAI(model="gpt-4-turbo")

# Define a clear and focused agent
planner_writer_agent = Agent(
    role='Article Planner and Writer',
    goal='Plan and then write a concise, engaging summary on a specified topic.',
    backstory=(
        'You are an expert technical writer and content strategist. '
        'Your strength lies in creating a clear, actionable plan before writing, '
        'ensuring the final summary is both informative and easy to digest.'
    ),
    verbose=True,
    allow_delegation=False,
    llm=llm
)

# Define a task with structured expected output
topic = "The importance of Reinforcement Learning in AI"
high_level_task = Task(
    description=(
        f"1. Create a bullet-point plan for a summary on the topic: '{topic}'.\n"
        f"2. Write the summary based on your plan, keeping it around 200 words."
    ),
    expected_output=(
        "A final report containing two distinct sections:\n\n"
        "### Plan\n"
        "- A bulleted list outlining the main points of the summary.\n\n"
        "### Summary\n"
        "- A concise and well-structured summary of the topic."
    ),
    agent=planner_writer_agent,
)

# Create the crew
crew = Crew(
    agents=[planner_writer_agent],
    tasks=[high_level_task],
    process=Process.sequential,
)

# Execute the task
print("## Running the planning and writing task ##")
result = crew.kickoff()

print("\n\n---\n## Task Result ##\n---")
print(result)
```

--------------------------------

### Main Function to Run Agent Queries Concurrently

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

The `main` asynchronous function sets up a list of tasks, each running `run_agent_with_tool` with a different query. It then uses `asyncio.gather` to execute these tasks concurrently, allowing multiple agent queries to be processed in parallel.

```python
async def main():
   """Runs all agent queries concurrently."""
   tasks = [
       run_agent_with_tool("What is the capital of France?"),
       run_agent_with_tool("What's the weather like in London?"),
       run_agent_with_tool("Tell me something about dogs.") # Should trigger the default tool response
   ]
   await asyncio.gather(*tasks)


nest_asyncio.apply()
asyncio.run(main())
```

--------------------------------

### Parallel Agent Execution for Web Research

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Demonstrates how to set up and use a ParallelAgent to run multiple LlmAgent instances concurrently. Each sub-agent is designed to research a specific topic using Google Search and output a concise summary.

```python
from google.adk.agents import LlmAgent, ParallelAgent
from google.adk.tools import google_search

GEMINI_MODEL = "gemini-2.0-flash-exp"

# Define Researcher Sub-Agents (to run in parallel)
researcher_agent_1 = LlmAgent(
    name="RenewableEnergyResearcher",
    model=GEMINI_MODEL,
    instruction="""
You are an AI Research Assistant specializing in energy.
Research the latest advancements in 'renewable energy sources'.
Use the Google Search tool provided.
Summarize your key findings concisely (1-2 sentences).
Output *only* the summary.""",
    description="Researches renewable energy sources.",
    tools=[google_search],
    output_key="renewable_energy_result"
)

researcher_agent_2 = LlmAgent(
    name="EVResearcher",
    model=GEMINI_MODEL,
    instruction="""
You are an AI Research Assistant specializing in transportation.
Research the latest developments in 'electric vehicle technology'.
Use the Google Search tool provided.
Summarize your key findings concisely (1-2 sentences).
Output *only* the summary.""",
    description="Researches electric vehicle technology.",
    tools=[google_search],
    output_key="ev_technology_result"
)

researcher_agent_3 = LlmAgent(
    name="CarbonCaptureResearcher",
    model=GEMINI_MODEL,
    instruction="""
You are an AI Research Assistant specializing in climate solutions.
Research the current state of 'carbon capture methods'.
Use the Google Search tool provided.
Summarize your key findings concisely (1-2 sentences).
Output *only* the summary.""",
    description="Researches carbon capture methods.",
    tools=[google_search],
    output_key="carbon_capture_result"
)

# Create the ParallelAgent
parallel_research_agent = ParallelAgent(
    name="ParallelWebResearchAgent",
    sub_agents=[researcher_agent_1, researcher_agent_2, researcher_agent_3],
    description="Runs multiple research agents in parallel to gather information."
)

```

--------------------------------

### Get Code Feedback from AI (Python)

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 11_ Goal Setting and Monitoring (Goal_Setting_Iteration).ipynb

Sends Python code and a list of goals to an AI language model for review. It generates a critique identifying goal attainment and areas for improvement.

```python
# Assuming 'llm' is an initialized language model object
def get_code_feedback(code: str, goals: list[str]) -> str:
    print("🔍 Evaluating code against the goals...")
    feedback_prompt = f"""
You are a Python code reviewer. A code snippet is shown below. Based on the following goals:

{chr(10).join(f"- {g.strip()}" for g in goals)}

Please critique this code and identify if the goals are met. Mention if improvements are needed for clarity, simplicity, correctness, edge case handling, or test coverage.

Code:
{code}
"""
    return llm.invoke(feedback_prompt)

```

--------------------------------

### Session Services for Memory Management in Python

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Illustrates different session service implementations for managing memory and state in an agentic application. It covers in-memory, database-backed, and Google Cloud Vertex AI-based session services.

```python
from google.adk.sessions import InMemorySessionService, DatabaseSessionService, VertexAiSessionService

# Example 1: Using InMemorySessionService
# Suitable for local development and testing
session_service = InMemorySessionService()

# Example 2: Using DatabaseSessionService
# Suitable for production with persistent storage
from google.adk.sessions import DatabaseSessionService

db_url = "sqlite:///./my_agent_data.db"
session_service = DatabaseSessionService(db_url=db_url)

# Example 3: Using VertexAiSessionService
# Suitable for scalable production on Google Cloud Platform
from google.adk.sessions import VertexAiSessionService

PROJECT_ID = "your-gcp-project-id"
LOCATION = "us-central1"
REASONING_ENGINE_APP_NAME = "projects/your-gcp-project-id/locations/us-central1/reasoningEngines/your-engine-id"

session_service = VertexAiSessionService(project=PROJECT_ID, location=LOCATION)

# When using VertexAI service, pass REASONING_ENGINE_APP_NAME to service methods:
# session_service.create_session(app_name=REASONING_ENGINE_APP_NAME, ...)
# session_service.get_session(app_name=REASONING_ENGINE_APP_NAME, ...)
# session_service.append_event(session, event, app_name=REASONING_ENGINE_APP_NAME)
```

--------------------------------

### Run Agent with Tool and Handle Execution

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This asynchronous Python function `run_agent_with_tool` takes a query string, invokes the agent executor, and prints the final output or any exceptions encountered during execution. It's designed to handle the invocation and display of results from the agent.

```python
async def run_agent_with_tool(query: str):
   """Invokes the agent executor with a query and prints the final response."""
   print(f"\n--- 🏃 Running Agent with Query: '{query}' ---")
   try:
       response = await agent_executor.ainvoke({"input": query})
       print("\n--- ✅ Final Agent Response ---")
       print(response["output"])
   except Exception as e:
       print(f"\n🛑 An error occurred during agent execution: {e}")
```

--------------------------------

### Indicate Task Completion by Agent

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This section confirms that a specific task, identified by a unique ID and assigned to an agent (e.g., 'Senior Financial Analyst'), has been successfully completed. It details the agent and any arguments used for the task.

```text
╭─╮╭─────────────────────────────────────────────── Task Completion ───────────────────────────────────────────────╮
│ │
│   Task Completed                                                                                                 │
│   Name: 3ae83180-2ac2-456f-aff6-2ff30a3e9f89                                                                     │
│   Agent: Senior Financial Analyst                                                                                │
│   Tool Args:                                                                                                     │
│ │
│ │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
```

--------------------------------

### Tool Use with Google Search using Google ADK

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Demonstrates how to integrate Google Search functionality into an AI agent using Google's ADK. It sets up an agent with the `google_search` tool and shows how to execute a search query through an asynchronous function. Requires ADK library and Google API access.

```python
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

# Define variables required for Session setup and Agent execution
APP_NAME = "Google_Search_agent"
USER_ID = "user1234"
SESSION_ID = "1234"

# Define Agent with access to search tool
root_agent = Agent(
    name="basic_search_agent",
    model="gemini-2.0-flash-exp",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[google_search]  # Google Search is a pre-built tool
)

# Agent Interaction
async def call_agent(query):
    """Helper function to call the agent with a query."""
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
    runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)

    content = types.Content(role='user', parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response:", final_response)

nest_asyncio.apply()
asyncio.run(call_agent("what's the latest ai news?"))
```

--------------------------------

### FastMCP Server for Model Context Protocol in Python

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Sets up a FastMCP server to expose a Python function as an Model Context Protocol (MCP) tool. This allows remote access to the 'greet' function, which takes a name and returns a greeting.

```python
from fastmcp import FastMCP, tool
import asyncio

# Define a simple tool function
# The @tool() decorator registers this Python function as an MCP tool
@tool()
def greet(name: str) -> str:
    """
    Generates a personalized greeting.

    Args:
        name: The name of the person to greet.

    Returns:
        A greeting string.
    """
    return f"Hello, {name}! Nice to meet you."

# Initialize the FastMCP server
# By default, FastMCP runs on http://localhost:8000
mcp_server = FastMCP()

if __name__ == "__main__":
    print("Starting FastMCP server...")
    print("This server exposes a 'greet' tool.")
    print("Access the tool schema at http://localhost:8000/tools.json")
    print("Press Ctrl+C to stop the server.")
    mcp_server.run()
```

--------------------------------

### Formulate and Run Financial Crew in Python

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Orchestrates the 'financial_analyst_agent' and 'analyze_aapl_task' into a 'Crew'. The crew is configured for verbose output. A main execution block includes a check for the 'OPENAI_API_KEY' environment variable before initiating the crew's execution via the 'kickoff' method.

```python
# --- 4. Formulate the Crew ---
# The crew orchestrates how the agent and task work together.
financial_crew = Crew(
 agents=[financial_analyst_agent],
 tasks=[analyze_aapl_task],
 verbose=True # Set to False for less detailed logs in production
)

```

## 4.3 Run Crew Agent

```python


# --- 5. Run the Crew within a Main Execution Block ---
# Using a __name__ == "__main__": block is a standard Python best practice.
def main():
   """Main function to run the crew."""
   # Check for API key before starting to avoid runtime errors.
   if not os.environ.get("OPENAI_API_KEY"):
       print("ERROR: The OPENAI_API_KEY environment variable is not set.")
       print("Please set it before running the script.")
       return


   print("\n## Starting the Financial Crew...")
   print("---------------------------------")

   # The kickoff method starts the execution.
   result = financial_crew.kickoff()


   print("\n---------------------------------")
   print("## Crew execution finished.")
   print("\nFinal Result:\n", result)


if __name__ == "__main__":
   main()

```
```

--------------------------------

### Sequential Agent Execution with Reflection Pattern

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Illustrates the Reflection Pattern using a SequentialAgent to chain two LlmAgents: a generator that creates initial content and a reviewer that fact-checks and critiques the generated content. This ensures a review process before final output.

```python
from google.adk.agents import SequentialAgent, LlmAgent

# The first agent generates the initial draft
generator = LlmAgent(
    name="DraftWriter",
    description="Generates initial draft content on a given subject.",
    instruction="Write a short, informative paragraph about the user's subject.",
    output_key="draft_text"
)

# The second agent critiques the draft
reviewer = LlmAgent(
    name="FactChecker",
    description="Reviews a given text for factual accuracy and provides a structured critique.",
    instruction="""
You are a meticulous fact-checker.
1. Read the text provided in the state key 'draft_text'.
2. Carefully verify the factual accuracy of all claims.
3. Your final output must be a dictionary containing two keys:
   - "status": A string, either "ACCURATE" or "INACCURATE".
   - "reasoning": A string providing a clear explanation for your status, citing specific issues if any are found.
""",
    output_key="review_output"
)

# SequentialAgent ensures the generator runs before the reviewer
review_pipeline = SequentialAgent(
    name="WriteAndReview_Pipeline",
    sub_agents=[generator, reviewer]
)

# Execution Flow:
# 1. generator runs -> saves paragraph to state['draft_text']
# 2. reviewer runs -> reads state['draft_text'] and saves dictionary to state['review_output']

```

--------------------------------

### Define Task for Stock Price Analysis in Python

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

Creates a 'Task' object for analyzing Apple's stock price. The task includes specific instructions for using the 'Stock Price Lookup Tool' and defines expected output formats for both successful retrieval and error cases. It assigns the 'financial_analyst_agent' to perform this task.

```python
analyze_aapl_task = Task(
 description=(
     "What is the current simulated stock price for Apple (ticker: AAPL)? "
     "Use the 'Stock Price Lookup Tool' to find it. "
     "If the ticker is not found, you must report that you were unable to retrieve the price."
 ),
 expected_output=(
     "A single, clear sentence stating the simulated stock price for AAPL. "
     "For example: 'The simulated stock price for AAPL is $178.15.' "
     "If the price cannot be found, state that clearly."
 ),
 agent=financial_analyst_agent,
)
```

--------------------------------

### Define Stock Price Lookup Tool

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This Python code defines a tool named `get_stock_price` that simulates fetching stock prices. It takes a ticker symbol and returns a float representing the price, or raises a ValueError if the ticker is not found in its simulated data. This tool is designed to be reusable and forces the agent to handle potential errors.

```python
# --- 1. Refactored Tool: Returns Clean Data ---
# The tool now returns raw data (a float) or raises a standard Python error.
# This makes it more reusable and forces the agent to handle outcomes properly.
@crew_tool("Stock Price Lookup Tool")
def get_stock_price(ticker: str) -> float:
   """
   Fetches the latest simulated stock price for a given stock ticker symbol.
   Returns the price as a float. Raises a ValueError if the ticker is not found.
   """
   logging.info(f"Tool Call: get_stock_price for ticker '{ticker}'")
   simulated_prices = {
       "AAPL": 178.15,
       "GOOGL": 1750.30,
       "MSFT": 425.50,
   }
   price = simulated_prices.get(ticker.upper())


   if price is not None:
       return price
   else:
       # Raising a specific error is better than returning a string.
       # The agent is equipped to handle exceptions and can decide on the next action.
       raise ValueError(f"Simulated price for ticker '{ticker.upper()}' not found.")
```

--------------------------------

### Implement Custom Agent by Extending BaseAgent

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 7_ Multi-Agent Collaboration - Code Example (ADK + Gemini Coordinator).ipynb

This snippet shows how to create a custom agent by inheriting from `BaseAgent`. It requires overriding the `_run_async_impl` method to define the agent's specific behavior, which does not involve an LLM. The example custom agent, `TaskExecutor`, yields a simple event to signify task completion.

```python
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

# Correctly implement a custom agent by extending BaseAgent
class TaskExecutor(BaseAgent):
    """A specialized agent with custom, non-LLM behavior."""
    name: str = "TaskExecutor"
    description: str = "Executes a predefined task."

    async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
        """Custom implementation logic for the task."""
        # This is where your custom logic would go.
        # For this example, we'll just yield a simple event.
        yield Event(author=self.name, content="Task finished successfully.")

```

--------------------------------

### Prompt Chaining with LangChain

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

Implements a prompt chaining pattern using LangChain Expression Language (LCEL). It defines two prompts for extracting technical specifications and transforming them into JSON. The chain processes input text, extracts information, and formats it as a JSON object. Requires LangChain and OpenAI libraries.

```python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Initialize the Language Model
llm = ChatOpenAI(temperature=0)

# Prompt 1: Extract Information
prompt_extract = ChatPromptTemplate.from_template(
    "Extract the technical specifications from the following text:\n\n{text_input}"
)

# Prompt 2: Transform to JSON
prompt_transform = ChatPromptTemplate.from_template(
    "Transform the following specifications into a JSON object with 'cpu', 'memory', and 'storage' as keys:\n\n{specifications}"
)

# Build the Chain using LCEL
extraction_chain = prompt_extract | llm | StrOutputParser()

# The full chain passes the output of the extraction chain
full_chain = (
    {"specifications": extraction_chain}
    | prompt_transform
    | llm
    | StrOutputParser()
)

# Run the Chain
input_text = "The new laptop model features a 3.5 GHz octa-core processor, 16GB of RAM, and a 1TB NVMe SSD."
final_result = full_chain.invoke({"text_input": input_text})

print("\n--- Final JSON Output ---")
print(final_result)
```

--------------------------------

### Create Parent Agent with Nested AgentTool (Python)

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Chapter 7_ Multi-Agent Collaboration - Code Example (ADK + Gemini AgentTooll).ipynb

Illustrates a parent agent, `Artist`, designed to utilize a nested `AgentTool` (`image_tool`). The `Artist` agent's instruction guides it to first create a descriptive prompt and then use the `ImageGen` tool (exposed via `image_tool`) to generate an image. This showcases agent composition.

```python
# 4. The parent agent remains unchanged. Its logic was correct.
artist_agent = LlmAgent(
    name="Artist",
    model="gemini-2.0-flash",
    instruction=(
        "You are a creative artist. First, invent a creative and descriptive prompt for an image. "
        "Then, use the `ImageGen` tool to generate the image using your prompt."
    ),
    tools=[image_tool]
)
```

--------------------------------

### Implement RAG with LangGraph and Weaviate

Source: https://context7.com/sarwarbeing-ai/agentic_design_patterns/llms.txt

This snippet sets up a Retrieval-Augmented Generation (RAG) system. It embeds and stores document chunks in Weaviate, defines a retriever, initializes an LLM, and builds a LangGraph to process user questions by retrieving relevant documents and generating concise answers. It requires the 'weaviate-client', 'langchain-openai', and 'langgraph' libraries.

```python
import weaviate
from weaviate.embedded_options import EmbeddedOptions
from langchain_community.vectorstores import Weaviate
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.schema import Document
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from typing import List, TypedDict
from langgraph.graph import StateGraph, END

# Embed and store chunks in Weaviate
client = weaviate.Client(embedded_options=EmbeddedOptions())

# Assume 'chunks' is a list of Document objects and 'retriever' is initialized
# For demonstration, let's mock these:
chunks = [Document(page_content="This is document 1."), Document(page_content="This is document 2.")]
# vectorstore = Weaviate.from_documents(
#     client=client,
#     documents=chunks,
#     embedding=OpenAIEmbeddings(),
#     by_text=False
# )
# retriever = vectorstore.as_retriever()

# Mock retriever for example purposes
class MockRetriever:
    def invoke(self, query):
        return [Document(page_content="Mock retrieved document content.")]
retriever = MockRetriever()

# Initialize LLM
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)

# Define the State for LangGraph
class RAGGraphState(TypedDict):
    question: str
    documents: List[Document]
    generation: str

# Define the Nodes (Functions)
def retrieve_documents_node(state: RAGGraphState) -> RAGGraphState:
    """Retrieves documents based on the user's question."""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question, "generation": ""}

def generate_response_node(state: RAGGraphState) -> RAGGraphState:
    """Generates a response using the LLM based on retrieved documents."""
    question = state["question"]
    documents = state["documents"]

    template = """You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise.
Question: {question}
Context: {context}
Answer:
"""
    prompt = ChatPromptTemplate.from_template(template)
    context = "\n\n".join([doc.page_content for doc in documents])
    rag_chain = prompt | llm | StrOutputParser()
    generation = rag_chain.invoke({"context": context, "question": question})
    return {"question": question, "documents": documents, "generation": generation}

# Build the LangGraph Graph
workflow = StateGraph(RAGGraphState)
workflow.add_node("retrieve", retrieve_documents_node)
workflow.add_node("generate", generate_response_node)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)

# Compile and run the graph
app = workflow.compile()

if __name__ == "__main__":
    print("\n--- Running RAG Query ---")
    query = "What did the president say about Justice Breyer"
    inputs = {"question": query}
    # Mocking the stream output as the actual execution requires API keys and setup
    print("Simulating app.stream output:")
    print({'retrieve': {'documents': [Document(page_content='Mock retrieved document content.')], 'question': query, 'generation': ''}})
    print({'generate': {'question': query, 'documents': [Document(page_content='Mock retrieved document content.')], 'generation': 'This is a mock generated response.'}})
    # for s in app.stream(inputs):
    #     print(s)

```

--------------------------------

### Define and Interact with a Basic Search Agent (Python)

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This snippet demonstrates how to define an agent named 'basic_search_agent' that uses the Google Search tool to answer user queries. It outlines the agent's configuration, including its name, model, description, instructions, and the tools it has access to. The interaction involves setting up a session, creating a runner, and executing the agent with a user query, printing the final response.

```python
from agentic_design_patterns import ADKAgent, Runner, InMemorySessionService, types
import asyncio
import nest_asyncio

# Assume google_search is a pre-defined tool
# Placeholder for demonstration purposes
class MockGoogleSearchTool:
    def __call__(self, query):
        return f"Mock search results for: {query}"

google_search = MockGoogleSearchTool()

APP_NAME = "my_app"
USER_ID = "user1"
SESSION_ID = "session1"

# Define Agent with access to search tool
root_agent = ADKAgent(
   name="basic_search_agent",
   model="gemini-2.0-flash-exp",
   description="Agent to answer questions using Google Search.",
   instruction="I can answer your questions by searching the internet. Just ask me anything!",
   tools=[google_search] # Google Search is a pre-built tool to perform Google searches.
)


# Agent Interaction
async def call_agent(query):
   """
   Helper function to call the agent with a query.
   """

   # Session and Runner
   session_service = InMemorySessionService()
   session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
   runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)

   content = types.Content(role='user', parts=[types.Part(text=query)])
   events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)


   for event in events:
       if event.is_final_response():
           final_response = event.content.parts[0].text
           print("Agent Response: ", final_response)

nest_asyncio.apply()
asyncio.run(call_agent("what's the latest ai news?"))
```

--------------------------------

### Final Crew Execution Summary

Source: https://github.com/sarwarbeing-ai/agentic_design_patterns/blob/main/notebooks/Copy of 5_tool_calling.ipynb

This is a summary output indicating the completion of crew execution. It provides the final result or conclusion derived from the collaborative work of the agents within the crew.

```text
Output:

---------------------------------
## Crew execution finished.

Final Result:
 The simulated stock price for AAPL is $178.15.


```