Google ADK + Pydantic AI — Multimodal ו-Type Safety

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import asyncio

# Minimal agent - no tools, just conversation
hello_agent = Agent(
    name="hello_agent",
    model="gemini-2.5-flash",
    instruction="You are a friendly assistant. Respond in Hebrew.",
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(
        agent=hello_agent,
        app_name="hello_app",
        session_service=session_service,
    )
    session = await session_service.create_session(
        app_name="hello_app", user_id="user_1",
    )
    message = types.Content(
        role="user",
        parts=[types.Part(text="שלום! מי אתה?")]
    )
    response = await runner.run_async(
        session_id=session.id,
        user_id="user_1",
        new_message=message,
    )
    for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)

asyncio.run(main())

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import asyncio

# --- Define a tool ---
def get_weather(city: str) -> dict:
    """Get the current weather for a city."""
    # In production, call a real weather API
    weather_data = {
        "Tel Aviv": {"temp": 28, "condition": "Sunny"},
        "Jerusalem": {"temp": 22, "condition": "Partly cloudy"},
        "Haifa": {"temp": 25, "condition": "Humid"},
    }
    return weather_data.get(city, {"temp": 20, "condition": "Unknown"})

# --- Define the agent ---
weather_agent = Agent(
    name="weather_agent",
    model="gemini-2.5-flash",
    instruction="""You are a helpful weather assistant.
    When asked about weather, use the get_weather tool.
    Always respond in the user's language.
    Be concise but friendly.""",
    tools=[get_weather],
)

# --- Run the agent ---
async def main():
    session_service = InMemorySessionService()
    runner = Runner(
        agent=weather_agent,
        app_name="weather_app",
        session_service=session_service,
    )

    session = await session_service.create_session(
        app_name="weather_app",
        user_id="user_1",
    )

    message = types.Content(
        role="user",
        parts=[types.Part(text="What's the weather in Tel Aviv?")]
    )

    response = await runner.run_async(
        session_id=session.id,
        user_id="user_1",
        new_message=message,
    )

    # Print the final response
    for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)

asyncio.run(main())

# Structure your project:
# my_agent/
#   __init__.py
#   agent.py  (must export `root_agent`)

# In agent.py:
from google.adk.agents import Agent

root_agent = Agent(
    name="my_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant.",
    tools=[],
)

# From terminal, in parent directory:
# adk web my_agent

from google.adk.agents import Agent
from google.genai import types
import base64
from pathlib import Path

# Agent that analyzes images
vision_agent = Agent(
    name="vision_agent",
    model="gemini-2.5-flash",
    instruction="""You are a visual analysis expert.
    When given an image, provide:
    1. A brief description of what you see
    2. Key objects and their positions
    3. Any text visible in the image
    4. Relevant insights for the user's question
    Be specific and factual. Do not hallucinate details.""",
)

# To send an image, include it in the Content:
async def analyze_image(runner, session, image_path: str, question: str):
    image_bytes = Path(image_path).read_bytes()

    message = types.Content(
        role="user",
        parts=[
            types.Part.from_image(
                image=types.Image(
                    image_bytes=image_bytes,
                    mime_type="image/jpeg",
                )
            ),
            types.Part(text=question),
        ],
    )

    response = await runner.run_async(
        session_id=session.id,
        user_id="user_1",
        new_message=message,
    )
    return response

from google.adk.agents import Agent
from google.adk.tools import google_search

research_agent = Agent(
    name="research_agent",
    model="gemini-2.5-flash",
    instruction="""You are a research assistant.
    Use Google Search to find current, accurate information.
    Always cite your sources.
    Focus on factual, verifiable claims.""",
    tools=[google_search],
)

# The agent will automatically search Google when
# it needs current information. Results come with
# citations and source URLs.

from typing import Optional

def search_products(
    query: str,
    category: Optional[str] = None,
    max_results: int = 5,
) -> list[dict]:
    """Search for products in the catalog.

    Args:
        query: Search keywords (e.g., "wireless headphones")
        category: Optional category filter (e.g., "electronics")
        max_results: Maximum number of results to return (default: 5)

    Returns:
        A list of matching products with name, price, and rating.
    """
    # Your product search logic here
    return [
        {"name": "Example Product", "price": 99.99, "rating": 4.5}
    ]

def calculate_discount(
    original_price: float,
    discount_percent: float,
) -> dict:
    """Calculate the discounted price.

    Args:
        original_price: The original price in dollars
        discount_percent: The discount percentage (0-100)

    Returns:
        Original price, discount amount, and final price.
    """
    discount_amount = original_price * (discount_percent / 100)
    final_price = original_price - discount_amount
    return {
        "original": original_price,
        "discount": round(discount_amount, 2),
        "final": round(final_price, 2),
    }

# Use in agent:
shop_agent = Agent(
    name="shop_agent",
    model="gemini-2.5-flash",
    instruction="You help customers find and compare products.",
    tools=[search_products, calculate_discount],
)

from google.adk.agents import Agent
from google.adk.tools import built_in_code_execution

analyst_agent = Agent(
    name="data_analyst",
    model="gemini-2.5-flash",
    instruction="""You are a data analyst.
    When asked to analyze data, write and execute Python code.
    Use matplotlib for charts, pandas for data manipulation.
    Always show your calculations.""",
    tools=[built_in_code_execution],
)

# The agent can now write and run Python code
# in a sandboxed environment, create charts,
# and return results.

from google.adk.agents import Agent
from google.adk.tools.mcp_tool import MCPToolset, SseServerParams

# Connect to an MCP server (e.g., filesystem server)
mcp_tools = MCPToolset(
    connection_params=SseServerParams(
        url="http://localhost:3001/sse",
    )
)

# Agent with both custom tools and MCP tools
file_agent = Agent(
    name="file_agent",
    model="gemini-2.5-flash",
    instruction="""You can read and analyze files.
    Use the available tools to access the filesystem.""",
    tools=[mcp_tools],
)

from google.adk.agents import Agent, SequentialAgent
from google.adk.tools import google_search

# Agent 1: Researcher
researcher = Agent(
    name="researcher",
    model="gemini-2.5-flash",
    instruction="""You are a research specialist.
    Search for information on the given topic.
    Write detailed research notes with sources.
    Save your findings to state['research_notes'].""",
    tools=[google_search],
)

# Agent 2: Analyst
analyst = Agent(
    name="analyst",
    model="gemini-2.5-flash",
    instruction="""You are a data analyst.
    Read the research notes from state['research_notes'].
    Identify key trends, statistics, and insights.
    Organize findings into a structured analysis.
    Save to state['analysis'].""",
)

# Agent 3: Writer
writer = Agent(
    name="writer",
    model="gemini-2.5-flash",
    instruction="""You are a professional content writer.
    Read the analysis from state['analysis'].
    Write a polished, engaging article (500-800 words).
    Include an executive summary, key findings, and conclusion.""",
)

# Pipeline: researcher -> analyst -> writer
research_pipeline = SequentialAgent(
    name="research_pipeline",
    sub_agents=[researcher, analyst, writer],
)

# This is now your root_agent:
root_agent = research_pipeline

from google.adk.agents import Agent, ParallelAgent, SequentialAgent

# Three agents that search different sources simultaneously
web_searcher = Agent(
    name="web_searcher",
    model="gemini-2.5-flash",
    instruction="Search the web for the given topic. Focus on recent news.",
    tools=[google_search],
)

academic_searcher = Agent(
    name="academic_searcher",
    model="gemini-2.5-flash",
    instruction="Find academic papers and research on the topic.",
)

market_searcher = Agent(
    name="market_searcher",
    model="gemini-2.5-flash",
    instruction="Find market data, company reports, and industry analysis.",
)

# Run all three searches simultaneously
parallel_search = ParallelAgent(
    name="parallel_search",
    sub_agents=[web_searcher, academic_searcher, market_searcher],
)

# Then synthesize results
synthesizer = Agent(
    name="synthesizer",
    model="gemini-2.5-flash",
    instruction="""Combine findings from all three searches.
    Create a comprehensive report with sections from each source.""",
)

# Full pipeline: parallel search -> synthesize
root_agent = SequentialAgent(
    name="full_pipeline",
    sub_agents=[parallel_search, synthesizer],
)

from google.adk.agents import Agent, LoopAgent

# Writer drafts content
drafter = Agent(
    name="drafter",
    model="gemini-2.5-flash",
    instruction="""Write or improve the current draft.
    If there is reviewer feedback in state, address it.
    Save your draft to state['current_draft'].""",
)

# Reviewer evaluates quality (1-10)
reviewer = Agent(
    name="reviewer",
    model="gemini-2.5-flash",
    instruction="""Review the draft in state['current_draft'].
    Score it 1-10 on clarity, accuracy, and engagement.
    If score >= 8, call the escalate tool to finish.
    If score < 8, provide specific improvement suggestions.""",
)

# Loop: write -> review -> repeat until quality >= 8
editing_loop = LoopAgent(
    name="editing_loop",
    sub_agents=[drafter, reviewer],
    max_iterations=3,  # Safety limit
)

from google.adk.agents import Agent
from google.adk.tools import ToolContext

def save_preference(
    key: str,
    value: str,
    tool_context: ToolContext,
) -> str:
    """Save a user preference to session state.

    Args:
        key: The preference name (e.g., "language", "format")
        value: The preference value (e.g., "Hebrew", "markdown")

    Returns:
        Confirmation message.
    """
    tool_context.state[key] = value
    return f"Saved {key} = {value}"

def get_preference(
    key: str,
    tool_context: ToolContext,
) -> str:
    """Get a user preference from session state.

    Args:
        key: The preference name to look up

    Returns:
        The preference value, or 'not set' if not found.
    """
    return tool_context.state.get(key, "not set")

preferences_agent = Agent(
    name="preferences_agent",
    model="gemini-2.5-flash",
    instruction="""You remember user preferences.
    Save preferences when the user tells you something.
    Use saved preferences to personalize responses.""",
    tools=[save_preference, get_preference],
)

from google.adk.sessions import DatabaseSessionService

# For production - sessions survive restarts
session_service = DatabaseSessionService(
    db_url="postgresql://user:pass@localhost/agents_db"
)

# Or with Firestore (Google Cloud native)
# session_service = FirestoreSessionService(
#     project_id="my-gcp-project",
#     collection="agent_sessions",
# )

# Use exactly like InMemorySessionService:
runner = Runner(
    agent=my_agent,
    app_name="production_app",
    session_service=session_service,
)

from google.adk.agents import Agent
from vertexai import agent_engines

# Define your agent as usual
my_agent = Agent(
    name="production_agent",
    model="gemini-2.5-flash",
    instruction="You are a production-ready assistant.",
    tools=[...],
)

# Deploy to Vertex AI Agent Engine
remote_agent = agent_engines.create(
    agent_engine=my_agent,
    requirements=["google-adk"],
    display_name="My Production Agent",
)

# Now query it via REST API or SDK
response = remote_agent.query(
    user_id="user_123",
    message="Hello!",
)
print(response)

# --- Pydantic v1 (old) ---
from pydantic import BaseModel, validator

class UserV1(BaseModel):
    name: str
    age: int

    @validator("age")
    def check_age(cls, v):
        if v < 0:
            raise ValueError("Age must be positive")
        return v

    class Config:
        orm_mode = True

# --- Pydantic v2 (current - required for Pydantic AI) ---
from pydantic import BaseModel, field_validator, ConfigDict

class UserV2(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # was orm_mode

    name: str
    age: int

    @field_validator("age")
    @classmethod
    def check_age(cls, v: int) -> int:
        if v < 0:
            raise ValueError("Age must be positive")
        return v

from pydantic_ai import Agent

# Simplest possible agent
simple_agent = Agent(
    model="openai:gpt-4o-mini",
    system_prompt="You are a helpful assistant. Be concise.",
)

# Synchronous usage (for scripts)
result = simple_agent.run_sync("What is the capital of Israel?")
print(result.output)  # "The capital of Israel is Jerusalem."

# Async usage (for production)
import asyncio

async def main():
    result = await simple_agent.run("What is the capital of Israel?")
    print(result.output)

asyncio.run(main())

from pydantic import BaseModel
from pydantic_ai import Agent

class CityInfo(BaseModel):
    """Information about a city."""
    name: str
    country: str
    population: int
    known_for: list[str]
    best_season_to_visit: str

# Agent with typed result
city_agent = Agent(
    model="openai:gpt-4o-mini",
    system_prompt="You are a travel expert. Provide accurate city information.",
    result_type=CityInfo,  # <- The magic!
)

result = city_agent.run_sync("Tell me about Tel Aviv")
city: CityInfo = result.output

# Now you have a fully typed object:
print(city.name)           # "Tel Aviv"
print(city.population)     # 460613
print(city.known_for)      # ["beaches", "nightlife", "startups", ...]
print(city.best_season_to_visit)  # "Spring (March-May)"

# Type checking works! Your IDE shows autocomplete.
# If the LLM returns invalid data, Pydantic validates
# and asks the LLM to try again automatically.

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from enum import Enum

class LeadScore(str, Enum):
    HOT = "hot"
    WARM = "warm"
    COLD = "cold"

class QualifiedLead(BaseModel):
    """Structured lead qualification result."""
    company_name: str
    score: LeadScore
    estimated_deal_size_usd: int = Field(ge=0)
    icp_match_percent: int = Field(ge=0, le=100)
    next_action: str
    reasoning: str
    disqualification_reason: str | None = None

lead_agent = Agent(
    model="openai:gpt-4o-mini",
    system_prompt="""You are a lead qualification specialist for
    an Israeli B2B SaaS company selling marketing automation.
    Our ICP: companies with 50-500 employees, B2B,
    tech/finance/healthcare, based in US/EU/IL.
    Qualify each lead as hot/warm/cold based on fit.""",
    result_type=QualifiedLead,
)

# Run qualification
result = lead_agent.run_sync(
    "New inbound lead: Sarah Chen, VP Marketing at HealthTech Inc. "
    "200 employees, Series B, San Francisco. Downloaded our ROI "
    "calculator and requested a demo."
)

lead = result.output
print(f"Score: {lead.score}")          # LeadScore.HOT
print(f"Deal size: ${lead.estimated_deal_size_usd:,}")
print(f"ICP match: {lead.icp_match_percent}%")
print(f"Next: {lead.next_action}")

# This is a real Python object - use it in your code:
if lead.score == LeadScore.HOT and lead.icp_match_percent > 80:
    # Auto-assign to senior AE, send to CRM
    assign_to_senior_rep(lead)
elif lead.score == LeadScore.COLD:
    # Add to nurture sequence
    add_to_nurture(lead)

from pydantic_ai import Agent

agent = Agent(
    model="openai:gpt-4o-mini",          # OpenAI
    # model="anthropic:claude-sonnet-4-20250514",  # Anthropic
    # model="google-gla:gemini-2.5-flash",       # Google
    # model="groq:llama-3.3-70b-versatile",   # Groq
    # model="ollama:llama3.2",                 # Local (Ollama)
    system_prompt="You are helpful.",
)

# Or override at runtime:
result = agent.run_sync(
    "Hello!",
    model="anthropic:claude-sonnet-4-20250514",  # Use Claude for this run
)

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

# --- Define dependencies ---
@dataclass
class MarketingDeps:
    """Dependencies injected into tools at runtime."""
    api_key: str
    company_name: str
    target_audience: str

# --- Define result type ---
class MarketingReport(BaseModel):
    """Structured marketing report."""
    company: str
    target_audience: str
    competitors: list[str]
    recommendations: list[str]
    estimated_budget_usd: int

# --- Create agent with deps type ---
marketing_agent = Agent(
    model="openai:gpt-4o",
    system_prompt="""You are a senior marketing strategist.
    Analyze the market and provide actionable recommendations.
    Use the tools available to gather data.""",
    deps_type=MarketingDeps,
    result_type=MarketingReport,
)

# --- Define tools with @agent.tool ---
@marketing_agent.tool
async def search_competitors(
    ctx: RunContext[MarketingDeps],
    industry: str,
) -> list[dict]:
    """Search for competitors in the given industry.

    Args:
        industry: The industry to search (e.g., "SaaS", "e-commerce")

    Returns:
        List of competitors with name and description.
    """
    # Access dependencies via ctx.deps
    company = ctx.deps.company_name

    # In production, call a real API
    return [
        {"name": "Competitor A", "description": f"Main rival to {company}"},
        {"name": "Competitor B", "description": "Rising challenger"},
    ]

@marketing_agent.tool
async def get_audience_insights(
    ctx: RunContext[MarketingDeps],
) -> dict:
    """Get insights about the target audience.

    Returns:
        Demographics, interests, and behavior patterns.
    """
    audience = ctx.deps.target_audience
    return {
        "segment": audience,
        "avg_age": 28,
        "interests": ["technology", "startups", "innovation"],
        "preferred_channels": ["LinkedIn", "Twitter", "YouTube"],
    }

# --- Run the agent ---
async def main():
    deps = MarketingDeps(
        api_key="sk-...",
        company_name="TechStartup IL",
        target_audience="Israeli tech professionals",
    )

    result = await marketing_agent.run(
        "Create a marketing strategy for our company",
        deps=deps,
    )

    report: MarketingReport = result.output
    print(f"Company: {report.company}")
    print(f"Budget: ${report.estimated_budget_usd}")
    for rec in report.recommendations:
        print(f"  - {rec}")

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    model="openai:gpt-4o-mini",
    deps_type=MarketingDeps,
    result_type=MarketingReport,
)

@agent.system_prompt
async def dynamic_prompt(ctx: RunContext[MarketingDeps]) -> str:
    """Generate system prompt based on dependencies."""
    return f"""You are a marketing strategist for {ctx.deps.company_name}.
    Today's date: {date.today()}.
    Target audience: {ctx.deps.target_audience}.
    Focus on the Israeli market and local competitors.
    Provide recommendations in both Hebrew and English."""

from pydantic_ai import Agent

agent = Agent(
    model="openai:gpt-4o-mini",
    system_prompt="You are a marketing copywriter. Write in Hebrew.",
)

async def stream_response():
    async with agent.run_stream("כתוב 3 כותרות לקמפיין השקה") as result:
        async for chunk in result.stream_text():
            print(chunk, end="", flush=True)
        # After stream completes, get usage info:
        final = await result.get_output()
        print(f"\n\nTokens used: {result.usage().total_tokens}")

from pydantic_ai import Agent
from pydantic_ai.exceptions import (
    UnexpectedModelBehavior,
    UserError,
)

agent = Agent(
    model="openai:gpt-4o-mini",
    system_prompt="You are helpful.",
    result_type=MyResultModel,
)

try:
    result = agent.run_sync("Analyze this lead")
except UnexpectedModelBehavior as e:
    # Model returned invalid output after max retries
    print(f"Model failed to produce valid output: {e}")
    # Fallback: use a different model or return default
    result = agent.run_sync(
        "Analyze this lead",
        model="anthropic:claude-sonnet-4-20250514",  # Try different model
    )
except UserError as e:
    # Configuration error (wrong model name, missing deps, etc.)
    print(f"Configuration error: {e}")
except Exception as e:
    # Network errors, API rate limits, etc.
    print(f"Unexpected error: {e}")