DEV Community: Claret Ibeawuchi

Callback the Police: Enforcing Business Rules in AI Agents 👮‍♂️

Claret Ibeawuchi — Wed, 29 Oct 2025 07:56:26 +0000

The Traffic Cop Analogy
What Are Callbacks in AI Agents?
The before_tool_callback: Your First Line of Defense
Real-World Pattern: Admin Payment Exemption
Real-World Pattern: Tool Interception and Routing
Real-World Pattern: Auto-Injecting Context
Real-World Pattern: Detecting Admin Message Leaks
Common Mistakes and How to Avoid Them
When to Use Which Callback
Key Takeaways
Resources

The Traffic Cop Analogy

Imagine a busy intersection without traffic lights or stop signs, just drivers making their own decisions. Most would probably stop and check. Some might slow down. A few might just go, assuming others will yield.

That's your AI agent without callbacks.

Now add a traffic cop who:

Blocks unsafe moves before they happen
Redirects traffic when routes are closed
Enforces rules consistently, regardless of driver intent

That's your AI agent with callbacks.

The difference? One relies on hoping drivers (the LLM) make good decisions. The other enforces good decisions with authority.

What Are Callbacks in AI Agents?

Callbacks are functions that run at specific points in your agent's lifecycle, giving you programmatic control over agent behavior.

Think of them as hooks where you can:

✅ Intercept actions before they execute
✅ Modify parameters or route to different functions
✅ Block unauthorized operations
✅ Inject additional context
✅ Validate outputs before users see them

The Callback Lifecycle

User: "Create an order for me"
         ↓
    [LLM reasoning...]
         ↓
    Decides to call: process_payment(amount=250)
         ↓
    ┌────────────────────────────────────┐
    │  before_tool_callback()            │  ← YOU ARE HERE
    │  "Should this be allowed?"         │    (Pre-execution)
    └────────────────────────────────────┘
         ↓
    [Tool executes: process_payment()]
         ↓
    [LLM generates response...]
         ↓
    ┌────────────────────────────────────┐
    │  after_model_callback()            │  ← YOU ARE HERE
    │  "Is this response correct?"       │    (Post-generation)
    └────────────────────────────────────┘
         ↓
    User sees: "Payment of $250 processed successfully"

Callbacks sit between the LLM's decisions and their execution/output, giving you the final say.

The before_tool_callback: Your First Line of Defense

The before_tool_callback runs before every tool call the LLM tries to make. This is where you enforce business rules.

Signature and Components

def before_tool_callback(
    tool: BaseTool,           # The tool about to be called
    args: dict,               # Arguments the LLM provided
    tool_context: ToolContext # Session state, auth token, etc.
) -> Optional[dict]:
    """
    Returns:
    - None: Proceed with normal tool execution
    - dict: Skip execution and return this result instead
    """
    pass

What You Can Do

def before_tool_callback(tool, args, tool_context):
    tool_name = tool.func.__name__

    # 1. BLOCK unauthorized actions
    if tool_name == "delete_user" and not is_admin(tool_context):
        return {
            "error": True,
            "message": "Unauthorized: Admin access required"
        }

    # 2. ROUTE to different implementations
    if tool_name == "get_price":
        if is_premium_tier(tool_context):
            return calculate_premium_price(**args)
        else:
            return calculate_standard_price(**args)

    # 3. MODIFY parameters before execution
    if tool_name == "search_database":
        # Add tenant filter for multi-tenant systems
        args["tenant_id"] = tool_context.state.get("tenant_id")

    # 4. INJECT missing context
    if tool_name == "create_order" and not args.get("user_id"):
        args["user_id"] = tool_context.state.get("current_user_id")

    return None  # Proceed with (possibly modified) execution

The callback can:

Return a result → Skip tool execution entirely, use your result
Modify args in place → Tool executes with modified parameters
Return None → Tool executes normally

Real-World Pattern: Admin Payment Exemption

Problem: Our agent was asking admin users to pay for services. Admins should create orders for free on behalf of customers.

Why it happened: The LLM doesn't understand business roles, it just follows the general user flow.

Solution: Intercept payment tool calls and block them for admin users.

Implementation

def before_tool_callback(tool, args, tool_context):
    tool_name = tool.func.__name__

    # Get or cache admin status
    is_admin = tool_context.state.get("session:is_admin")

    if is_admin is None:
        # First time - check auth token and cache result
        auth_token = tool_context.state.get("auth_token")
        is_admin = check_admin_status(auth_token)
        tool_context.state["session:is_admin"] = is_admin  # Cache it

    # ========================================================================
    # ADMIN PAYMENT EXEMPTION
    # Block payment tools for admin users - admins create orders for free
    # ========================================================================
    if tool_name in ["process_payment", "charge_card"]:
        if is_admin:
            logger.info(f"🔐 ADMIN USER: Blocking {tool_name} - no payment required")

            # Mark session as payment exempt
            tool_context.state["session:payment_status"] = "admin_exempt"

            # Return success WITHOUT executing payment
            return {
                "success": True,
                "message": "✅ Admin user - no payment required. You can proceed immediately.",
                "admin_exemption": True,
                "payment_status": "admin_exempt"
            }

    return None  # Non-payment tools execute normally

What Happens

Before callbacks:

Admin: "Create an order"
Agent: "I'll need to process your payment first..."
Admin: /frustration/ "I'm an admin!"

After callbacks:

Admin: "Create an order"
[Callback intercepts payment tool call]
[Returns success without executing payment]
Agent: "✅ Admin user - no payment required. Creating your order now..."
Admin: /happy/

Key Insights

Admin status is cached in session state, we only check once per session
Multiple payment tool names are covered (any payment-related tool)
Logs include context (🔐 emoji for security actions)
Session state updated to track exemption status
Clean response that explains why (for debugging) without confusing the user

Performance Optimization: Caching Expensive Checks

def before_tool_callback(tool, args, tool_context):
    # Check cache first
    is_admin = tool_context.state.get("session:is_admin")

    if is_admin is None:
        # First time - check auth and cache result
        auth_token = tool_context.state.get("auth_token")
        is_admin = check_admin_status(auth_token)  # Expensive: JWT decode + lookup

        # Cache for entire session
        tool_context.state["session:is_admin"] = is_admin
        logger.info(f"🔐 Cached admin status: {is_admin}")

    # Use cached value
    if tool_name in PAYMENT_TOOLS and is_admin:
        return {"success": True, "admin_exempt": True}

    return None

Performance impact:

Before: Auth check on every tool call (~50-100ms each × 10 calls = 500-1000ms)
After: Auth check once per session (~50ms total)

What to cache:

✅ User roles/permissions
✅ Tenant/organization IDs
✅ Feature flags
❌ Dynamic data (current balance, order status)

Real-World Pattern: Tool Interception and Routing

Problem: The LLM sometimes calls the wrong pricing tool:

Calls regular pricing for admin users → Wrong (too expensive)
Calls admin pricing for regular users → Security violation

Why it happened: The LLM chooses tools based on context and phrasing, which is inconsistent.

Solution: Intercept pricing tool calls and route to the correct implementation.

Implementation

def before_tool_callback(tool, args, tool_context):
    tool_name = tool.func.__name__

    # Get user type from session
    is_admin = tool_context.state.get("session:is_admin", False)

    # ========================================================================
    # DETERMINISTIC PRICING ROUTING
    # If LLM calls wrong pricing tool, redirect to correct one
    # ========================================================================

    if tool_name == "get_standard_price" and is_admin:
        # Admin called standard pricing → Route to premium
        logger.warning(
            f"🔄 LLM called standard pricing for ADMIN user - "
            f"intercepting and routing to premium pricing"
        )

        # Import and call correct function
        from app.tools.pricing import get_premium_price

        # Pass through all args + context
        args["tool_context"] = tool_context
        return get_premium_price(**args)

    elif tool_name == "get_premium_price" and not is_admin:
        # Regular user called admin pricing → Security violation, route to standard
        logger.warning(
            f"🔄 LLM called premium pricing for REGULAR user - "
            f"intercepting and routing to standard pricing"
        )

        from app.tools.pricing import get_standard_price
        args["tool_context"] = tool_context
        return get_standard_price(**args)

    return None

What Happens

Regular User: "How much for this service?"
LLM: [Decides to call get_premium_price() - WRONG TOOL]
      ↓
[Callback intercepts]
      ↓
[Routes to get_standard_price() instead]
      ↓
User sees correct standard pricing ✅

Logs show:
"🔄 LLM called premium pricing for REGULAR user - intercepting and routing"

Benefits

Zero reliance on LLM choosing the right tool
Security enforced in code, not prompts
Transparent logging of routing decisions
No user-visible errors - routing happens silently
Consistent behavior regardless of how user phrases request

Real-World Pattern: Auto-Injecting Context

Problem: After getting a quote, users would say "proceed with order" and the agent would re-ask for all the parameters, even though all that information was already collected.

Why it happened: The LLM didn't pass previous context to the order creation tool.

Solution: Auto-inject stored quote details when order tools are called.

Implementation

def before_tool_callback(tool, args, tool_context):
    tool_name = tool.func.__name__

    # ========================================================================
    # AUTO-INJECT QUOTE DETAILS INTO ORDER
    # If order tool is called without params, fill from stored quote
    # ========================================================================

    if tool_name == "create_order":
        # Get previously stored quote from session
        latest_quote = tool_context.state.get("session:latest_quote")

        if latest_quote:
            # Auto-inject missing required parameters
            for key in ["param_a", "param_b", "param_c", "total_price"]:
                if not args.get(key) and latest_quote.get(key):
                    args[key] = latest_quote[key]
                    logger.info(f"✅ Auto-injected {key}: {args[key]}")

    return None  # Proceed with augmented args

Storing Quote Data

When a quote is generated, store it:

def calculate_quote(params, tool_context):
    # Calculate quote
    price = api.get_price(params)

    # Store complete quote in session state
    quote_data = {
        "param_a": params.get("param_a"),
        "param_b": params.get("param_b"),
        "param_c": params.get("param_c"),
        "total_price": price,
        "timestamp": time.time()
    }

    tool_context.state["session:latest_quote"] = quote_data

    return {"price": price, ...}

User Experience

Before callbacks:

User: "Quote for service from A to B"
Agent: "That'll be $1,200"
User: "Book it"
Agent: "I'll need parameter A"
User: "I just told you!"
Agent: "I need the exact value..."

After callbacks:

User: "Quote for service from A to B"
Agent: "That'll be $1,200"
[Quote stored in session state]
User: "Book it"
[Callback injects all required params from session]
Agent: "Great! Creating your order now..."

Logs show:

✅ Auto-injected param_a: value_a
✅ Auto-injected param_b: value_b
✅ Auto-injected param_c: value_c
✅ Auto-injected price: $1200.00

Key Insights

Session state bridges turns - Data persists across messages
LLM doesn't need to remember - We do it for them
Only inject if missing - Don't override if LLM provided it
Log every injection - Helps debugging and monitoring
Structured storage - Use consistent keys for retrieval

Common Mistakes and How to Avoid Them

❌ Mistake 1: Not Returning Anything

def before_tool_callback(tool, args, tool_context):
    if should_block_tool(tool):
        logger.error("Blocking tool")
        # BUG: Forgot to return! Tool will still execute

Fix:

def before_tool_callback(tool, args, tool_context):
    if should_block_tool(tool):
        logger.error("Blocking tool")
        return {"error": True, "message": "Unauthorized"}  # ✅

❌ Mistake 2: Modifying Args Without In-Place Updates

def before_tool_callback(tool, args, tool_context):
    # BUG: Creates new dict, doesn't modify original
    args = {**args, "tenant_id": "abc123"}
    return None

Fix:

def before_tool_callback(tool, args, tool_context):
    # ✅ Modifies dict in-place
    args["tenant_id"] = "abc123"
    return None

❌ Mistake 3: Forgetting Tool Context Parameter

# Original tool signature
def my_tool(param1, param2):
    pass

# In callback - BUG: Doesn't pass tool_context
return my_other_tool(param1=args["param1"])

Fix:

# If tool needs context, include it
return my_other_tool(
    param1=args["param1"],
    tool_context=tool_context  # ✅
)

❌ Mistake 4: Missing Tool Name Variants

def before_tool_callback(tool, args, tool_context):
    # BUG: Only catches one tool name, misses variants
    if tool.func.__name__ == "charge_payment":
        if is_admin(tool_context):
            return {"success": True, "admin_exempt": True}

Problem: Your system might have multiple payment-related tools:

charge_payment
initiate_payment
process_card_payment

Missing one means the callback doesn't catch it!

Fix:

# Define at module level for easy maintenance
PAYMENT_TOOLS = [
    "charge_payment",
    "initiate_payment", 
    "process_card_payment",
    "charge_card"
]

def before_tool_callback(tool, args, tool_context):
    tool_name = tool.func.__name__

    # ✅ Catches all variants
    if tool_name in PAYMENT_TOOLS:
        if is_admin(tool_context):
            return {"success": True, "admin_exempt": True}

    return None

Maintenance tip: When adding new tools, update the list once. All callbacks automatically cover it.

❌ Mistake 5: Catching All Tools Too Broadly

def before_tool_callback(tool, args, tool_context):
    # BUG: Every tool will get user_id added
    args["user_id"] = get_user_id(tool_context)
    return None

Fix:

def before_tool_callback(tool, args, tool_context):
    # ✅ Only inject for tools that need it
    if tool.func.__name__ in ["create_order", "update_profile"]:
        args["user_id"] = get_user_id(tool_context)
    return None

❌ Mistake 6: No Logging

def before_tool_callback(tool, args, tool_context):
    if is_admin(tool_context):
        return admin_version(**args)
    return None

Fix:

def before_tool_callback(tool, args, tool_context):
    if is_admin(tool_context):
        logger.info(f"🔄 Routing {tool.func.__name__} to admin version")
        return admin_version(**args)
    return None

Logging is critical for debugging "why did this happen?" in production.

Real-World Pattern: Detecting Admin Message Leaks

Even with perfect tool routing and parameter injection, the LLM can still leak admin-only information in its natural language responses.

The Problem

Admin sees:
"Using admin access for premium pricing... Your cost: $1,200"

Non-admin sees:
"Using admin access for premium pricing... Total: $1,500" ❌

The tool correctly returned different prices, but the LLM copy-pasted admin messaging into non-admin responses!

The Solution: Response Content Validation

Use after_model_callback to detect and clean role-specific messaging:

def after_model_callback(callback_context, llm_response):
    """Detect and remove admin-only phrases from non-admin responses"""

    # Get user role from session
    is_admin = callback_context.state.get("session:is_admin", False)

    if is_admin:
        return None  # Admin can see admin messages

    # Extract response text
    response_text = llm_response.content.parts[0].text

    # Define admin-only phrases
    admin_only_phrases = [
        "using admin access",
        "premium pricing",
        "admin access for",
        "your premium cost"
    ]

    # Check if any admin phrase leaked
    has_admin_phrase = any(
        phrase.lower() in response_text.lower() 
        for phrase in admin_only_phrases
    )

    if has_admin_phrase:
        logger.warning(
            f"🚨 Admin message leaked to non-admin user! "
            f"Cleaning response..."
        )

        # Get correct price from session
        display_price = callback_context.state.get("session:last_display_price")

        # Override with clean, role-appropriate message
        clean_response = f"Total shipping cost: ${display_price:.2f} 🚚"

        from google.genai import types
        return LlmResponse(
            content=types.Content(
                role="model",
                parts=[types.Part(text=clean_response)]
            )
        )

    return None  # Response is clean

Why This Happens

LLMs learn patterns from context. If they recently saw:

Admin messages in earlier conversation
Admin-formatted responses in training data
Tool responses that mention "premium" or "admin"

They might echo these patterns even for non-admin users.

When to Use This Pattern

✅ Use when:

Different user roles see different information
LLM has access to role-specific tool responses
Sensitive information must never leak

❌ Don't use when:

All users see the same information
No role-based differentiation needed

Lesson learned: Callbacks protect against both wrong tool execution AND wrong LLM messaging. Defense in depth applies to language, not just code.

When to Use Which Callback

Google ADK provides several callback types. Here's when to use each:

`before_tool_callback` - Pre-Execution Enforcement

Use when:

✅ Blocking unauthorized actions
✅ Routing to different implementations
✅ Injecting authentication context
✅ Auto-filling parameters from session
✅ Validating parameters before execution

Example use cases:

Admin payment exemption
Multi-tenant data isolation
Rate limiting
Parameter validation

`after_model_callback` - Post-Generation Validation

Use when:

✅ Validating LLM output accuracy
✅ Sanitizing responses (removing sensitive data)
✅ Formatting output consistently
✅ Correcting known LLM mistakes

Example use cases:

Price validation (next article!)
PII detection and removal
Response format enforcement
Output translation

Comparison

Callback Type	When It Runs	What to Use It For
`before_tool_callback`	Before tool execution	Business rule enforcement, routing, injection
`after_tool_callback`	After tool execution	Result transformation, logging
`after_model_callback`	After LLM generates response	Output validation, correction

Most common pattern: Use before_tool_callback for enforcement and after_model_callback for validation.

Key Takeaways

Callbacks are enforcers - They don't ask, they tell
Use them for business logic - Don't rely on LLM prompt-following
Log everything - You'll need it for debugging
Cache expensive checks - Store results in session state
Return early for blocks - Don't execute what shouldn't happen
Modify args in-place - When you want execution to proceed with changes
Test edge cases - Callbacks run on EVERY tool call

The Enforcement Hierarchy

Strongest:   Code enforcement (callbacks)
             ↑
Medium:      Tool filtering by role
             ↑
Weakest:     LLM instructions

Put your critical business logic in callbacks, not instructions.

Resources

Previous: "How Reliable Are Your AI Agents?"

Next: "The Ground Truth Principle: Session State and Output Validation"

What business rules are you enforcing with callbacks? Share your patterns!

How Reliable Are Your AI Agents?

Claret Ibeawuchi — Wed, 29 Oct 2025 07:21:56 +0000

The $117 Bug That Changed Everything
The Reliability Gap
Why LLMs Fail at Critical Tasks
The Multi-Layer Defense Strategy
Real-World Impact
What's Next
Resources

The $117 Bug That Made Me Rethink Architecture

Picture this: A customer requests a shipping quote. Your AI agent calculates the price: $1,320.56. The agent stores it correctly in the database. The backend logs show the right number. Everything looks perfect.

But when the customer sees the response, it says: $117.00.

That's a 91% error. One hallucination away from massive financial loss or customer distrust.

This actually happened in production. And it's not an edge case, it's a fundamental characteristic of Large Language Models.

The Reliability Gap

When we talk about "AI agents," we're usually talking about LLMs with access to tools (functions they can call). The promise is compelling: natural language interfaces that can handle complex business logic.

The reality? LLMs are very advanced prediction engines, not calculators.

What LLMs Are Good At

Understanding natural language and intent
Generating human-like, contextual responses
Recognizing patterns and relationships
Adapting to conversational flow

What LLMs Struggle With

Exact calculations and number precision
Following strict business rules consistently
Deterministic behavior across requests
Security boundaries and access control
Critical decision-making without oversight

The Developer's Dilemma

Think of it like hiring someone who's brilliant at customer service but occasionally forgets basic math:

# This is what you write
def calculate_price(user_type, base_price):
    if user_type == "wholesale":
        return base_price * 0.85  # 15% discount
    else:
        return base_price * 1.12  # 12% markup

# This is what the LLM might do
"Based on the pricing, I calculate roughly $117 for this service..."

The LLM might:

Use the wrong pricing tier (wholesale vs retail)
Misread numbers during token processing (1320.56 → $117.00)
Hallucinate calculations instead of using tool results
Ignore access controls when constructing responses
Mix data from different contexts

You can't fix this with better prompts alone. This requires architectural solutions.

Why LLMs Fail at Critical Tasks

1. Token-Based Processing

LLMs don't process numbers as mathematical values, they process everything as tokens (text chunks).

The number 1320.56 might be tokenized as:

["1320", ".", "56"]
["13", "20", ".56"]
["1", "320", ".", "56"]

Depending on the tokenizer. When generating output, the model predicts the most statistically likely next token, not the mathematically correct one.

2. Probabilistic, Not Deterministic

# Deterministic code - same input = same output
def check_permission(user):
    return user.role == "admin"  # Always consistent

# LLM reasoning - same input ≠ guaranteed same output
"""
The user appears to have administrative permissions based on 
their previous actions... (might vary across calls)
"""

With temperature > 0, the same prompt can yield different outputs. Even at temperature 0, subtle prompt changes can shift behavior.

3. Context Window Limitations

Your agent might have the correct data in its context, but:

As conversations grow, early information gets less attention
Critical details might be overshadowed by recent messages
The model might "forget" information from 50+ messages ago

4. No Inherent Security Model

LLMs don't understand:

Role-based access control (RBAC)
Authentication vs Authorization
Privilege escalation risks
Data isolation requirements

They might cheerfully expose wholesale pricing to retail customers if the context suggests it, or process admin commands from regular users.

5. Instruction Drift

Even with perfect instructions, LLMs can:

Misinterpret edge cases
Prioritize recent conversation over system instructions
Follow user instructions that contradict system rules
Generate plausible-sounding but incorrect responses

The Multi-Layer Defense Strategy

Here's what I learned building production AI agents: Don't trust the LLM for business-critical logic. Use it for what it does best, and enforce everything else deterministically.

Think of it like airport security—multiple checkpoints, each catching different issues:

The Architecture

User Request
     ↓
┌─────────────────────────────────────────────┐
│  Layer 1: Tool Filtering                    │  ← "What can this user access?"
│  (Before agent even sees the tools)         │
└─────────────────────────────────────────────┘
     ↓
┌─────────────────────────────────────────────┐
│  Layer 2: Instruction Engineering           │  ← "How should you behave?"
│  (Guide LLM with clear rules)               │
└─────────────────────────────────────────────┘
     ↓
┌─────────────────────────────────────────────┐
│  Layer 3: Pre-Execution Callbacks           │  ← "Is this allowed right now?"
│  (before_tool_callback - Business rules)    │
└─────────────────────────────────────────────┘
     ↓
┌─────────────────────────────────────────────┐
│  Layer 4: Tool Execution                    │  ← "Execute with ground truth"
│  (Deterministic business logic)             │
└─────────────────────────────────────────────┘
     ↓
┌─────────────────────────────────────────────┐
│  Layer 5: Output Validation                 │  ← "Is the response accurate?"
│  (after_model_callback - Final check)       │
└─────────────────────────────────────────────┘
     ↓
User Response

Each layer provides a different type of safety:

Layer 1 (Tool Filtering): "What tools can this user even see?"
Layer 2 (Instructions): "How should the LLM behave?"
Layer 3 (Pre-Execution): "Is this specific tool call allowed? Should we route it elsewhere?"
Layer 4 (Execution): "Calculate with code, not with tokens"
Layer 5 (Validation): "Is the output what we expected? Correct if needed"

Example: Catching the $117 Bug

Here's how the layers work together:

# Layer 4: Tool execution - Calculate with code (ground truth)
def calculate_shipping_price(origin, destination, vehicle):
    price = pricing_api.get_quote(origin, destination, vehicle)
    # price = 1320.56 (calculated correctly)

    # Store ground truth in session state
    session.state['expected_price'] = price

    return {
        "total_price": price,
        "currency": "USD",
        "breakdown": {...}
    }

# Layer 5: Validation - Catch LLM errors before user sees them
def validate_response(context, llm_response):
    """
    This runs AFTER the LLM generates a response but BEFORE 
    the user sees it. It's our last line of defense.
    """
    import re

    # What did the LLM say?
    response_text = llm_response.content.parts[0].text
    mentioned_prices = re.findall(r'\$\s*([0-9,]+\.?\d*)', response_text)

    if not mentioned_prices:
        return None  # No prices mentioned, all good

    # What should it have said?
    expected_price = context.state.get('expected_price')  # 1320.56
    mentioned_price = float(mentioned_prices[0].replace(',', ''))  # 117.00

    # Validate within tolerance (2% for rounding)
    difference_pct = abs(mentioned_price - expected_price) / expected_price

    if difference_pct > 0.02:  # More than 2% off
        logger.error(
            f"🚨 PRICE ERROR: LLM showed ${mentioned_price}, "
            f"expected ${expected_price} (Δ {difference_pct*100:.1f}%)"
        )

        # Correct it silently - user never knows
        return LlmResponse(
            content=Content(
                role="model",
                parts=[Part(text=f"Total shipping cost: ${expected_price:.2f} 🚚")]
            )
        )

    return None  # Price is correct, use original response

What happened in my case:

✅ Tool calculated correct price: $1,320.56
✅ Stored as ground truth in session
❌ LLM generated response with wrong price: $117.00 (91% error!)
✅ Validation caught the discrepancy
✅ Response corrected to $1,320.56 before user saw it
✅ Error logged for monitoring and debugging

Customer saw: "Total shipping cost: $1,320.56" ✅

Logs showed: "🚨 PRICE ERROR #1 DETECTED" 📊

Real-World Impact

Let's compare approaches:

Before: Hope and Pray 🙏

# Traditional approach - trust the LLM completely
agent = LLM(
    tools=[calculate_price, charge_card, book_shipment],
    instructions="""
    You are a helpful shipping assistant. 
    Always use the calculate_price tool for quotes.
    Be accurate with prices.
    Don't charge admin users.
    """
)

# Cross fingers and hope it follows instructions...

Observed failure rates:

Wrong pricing tier: ~15-20% of requests
Hallucinated numbers: ~5-10% of price displays
Admin privilege leaks: ~3-5% of admin sessions
Inconsistent behavior: Varies with prompt changes

After: Defense in Depth 🛡️

# Production-ready approach - enforce with code
agent = LLM(
    # Layer 1: Filter tools by user role
    tools=get_allowed_tools_for_user(user),

    # Layer 2: Context-aware instructions
    instructions=build_instructions(user_context),

    # Layer 3: Pre-execution enforcement
    before_tool_callback=enforce_business_rules,

    # Layer 5: Post-generation validation
    after_model_callback=validate_output
)

def enforce_business_rules(tool, args, context):
    """Runs before EVERY tool call"""
    user_role = context.state.get('user_role')

    # Block unauthorized actions
    if tool.name == 'charge_card' and user_role == 'admin':
        logger.info("Admin users don't pay - blocking payment tool")
        return {
            "success": True,
            "message": "Admin booking confirmed - no payment required",
            "skipped": True
        }

    # Route to correct implementation
    if tool.name == 'calculate_price':
        if user_role == 'admin':
            # Route to wholesale pricing
            return calculate_wholesale_price(args, context)
        else:
            # Route to retail pricing
            return calculate_retail_price(args, context)

    return None  # Proceed with normal execution

def validate_output(context, response):
    """Runs after LLM generates response"""
    # Check prices match ground truth
    # Check no sensitive data leaked
    # Check response format is correct
    # Return corrected response if needed
    pass

Results after implementation:

✅ Wrong pricing tier: 0% (blocked by callback)
✅ Hallucinated numbers: 0% (caught by validation)
✅ Admin privilege leaks: 0% (filtered at tool level)
✅ Inconsistent behavior: Eliminated (enforced by code)
⚠️ False positives: <0.1% (logged, reviewed, tuned)

The Reliability Equation

Agent Reliability = 
    MIN(
        LLM Accuracy,
        Instruction Following Rate,
        Context Retention
    )
    +
    Callback Coverage × Enforcement Quality
    +
    Validation Coverage × Error Detection Rate

The first line is unpredictable and varies with:

Model updates
Prompt changes
Conversation length
User phrasing

The second and third lines are deterministic—they're code you control.

Don't bet your business on only the first line.

What's Next

Building reliable AI agents isn't about making the LLM perfect, it's about designing systems that deliver reliable results despite LLM imperfections.

In this series, we'll explore each defensive layer:

Article 2: "Callback the Police" 👮‍♂️

How to use callbacks to enforce business rules like law enforcement—stopping bad behavior before it happens.

Article 3: "The Ground Truth Principle" 📊

Why session state is your source of truth and how to use it for validation.

Article 4: "Explicit Contracts Save Lives" ⚖️

Making function parameters explicit so callbacks can inject and enforce them.

Article 5: "Event-Driven Automation" 🔄

Using webhooks and background tasks to make agents proactive and reliable.

Try It Yourself

Quick reliability audit for your agent:

Question 1: If the LLM calls the wrong tool, do you detect and block it?

❌ No → Implement before_tool_callback for routing
⚠️ Sometimes → Ensure callbacks cover all tools
✅ Yes → Great! Is your coverage monitored?

Question 2: If the LLM displays incorrect data, do you catch it?

❌ No → Implement after_model_callback for validation
⚠️ Log only → Add correction logic
✅ Yes and correct → Excellent!

Question 3: Can the LLM bypass access controls?

❌ Yes → Filter tools by user role
⚠️ Depends on prompts → Make it code-enforced
✅ No → Verify with penetration testing

Question 4: Do you validate critical outputs?

❌ No → Start with financial/legal data
⚠️ Some → Expand to all critical outputs
✅ All → Document validation coverage

Question 5: Can you debug what went wrong?

❌ No logs → Add structured logging
⚠️ Basic logs → Add context and tracing
✅ Full tracing → Can you alert on patterns?

If you answered ❌ or ⚠️ to any question, you need deterministic layers. Your agent is relying too much on LLM behavior.

Key Takeaways

LLMs are amazing at conversation, terrible at guarantees - Use them for what they do best
Multiple defense layers - Like airport security, assume each layer might miss something
Ground truth in code - Store correct values, validate against them
Callbacks are enforcement - They're your business logic police
Monitor everything - You can't improve what you don't measure

Resources

Next Article: "Callback the Police: Enforcing Business Rules in AI Agents" →

Have you caught reliability issues in your AI agents? What strategies worked for you? Share in the comments!

Understanding Google Maps Grounding with ADK (Part 2/5)

Claret Ibeawuchi — Sat, 04 Oct 2025 08:59:37 +0000

📚 This is Part 2 of a 5-part series on building production-ready AI agents with Google's Agent Development Kit (ADK):

✅ Part 1: Google Search Grounding with ADK (complete this first!)

📍 Part 2: Understanding Google Maps Grounding with ADK (you are here)

Part 3: Building a Full-Stack Frontend with CopilotKit & AG-UI

Part 4: Persistent Sessions with PostgreSQL & Docker

Part 5: Production Deployment on Cloud Run

⚠️ Important: If you haven't completed Part 1, start there first! This tutorial builds directly on the concepts and setup from Part 1.

Introduction

Welcome back! In Part 1, we built an AI agent with Google Search grounding that can access real-time web information. Now in Part 2, we're adding location intelligence with Google Maps grounding.

Google Maps Grounding is a powerful feature in the Agent Development Kit (ADK) that enables AI agents to access information from Google Maps' vast database of places, businesses, and location data. By connecting your agents to Google Maps, you can provide users with accurate, location-aware responses grounded in real-world place information.

This feature is particularly valuable for queries requiring information about:

🏢 Businesses, restaurants, and services
📍 Addresses and locations
⭐ Ratings, reviews, and hours of operation
🚗 Directions and distance information

When your agent determines that location-based information is needed, it automatically searches Google Maps and incorporates the results into its response with proper attribution.

The Challenge: Limited Documentation

In Part 1, we had solid documentation for Search grounding. Part 2 is different. Unlike Vertex AI Search grounding, which has comprehensive documentation, Google Maps grounding in ADK is still evolving. When I built this integration, I had:

Google Cloud's Vertex AI Maps Grounding docs - for the underlying API
The ADK Python source code - the implementation
No working ADK-specific examples or quickstart guides

I spent hours debugging, reading source code, and testing different approaches. This article documents what I learned so you can skip the struggle and use this as the Maps grounding documentation that doesn't yet exist officially.

What You'll Learn

Building on the foundation from Part 1, in this guide you'll discover:

Quick Setup: How to create and run a Google Maps-enabled agent from scratch
Grounding Architecture: The data flow and technical process behind Maps grounding
Response Structure: How to interpret grounded responses and their metadata
Current Limitations: Understanding what Maps grounding can and cannot do (yet)
Workarounds: The Agent-as-Tool pattern that makes everything work reliably
Best Practices: Guidelines for production implementations

Google Maps Grounding Quickstart

This quickstart guides you through creating an ADK agent with Google Maps grounding capability. This quickstart assumes a local IDE (VS Code or PyCharm, etc.) with Python 3.9+ and terminal access.

Prerequisites

Before starting Part 2, ensure you have:

✅ Completed Part 1 - Search grounding agent working

✅ Google Cloud Project - With billing enabled

✅ Vertex AI API enabled - This is critical for Maps grounding

✅ Google Cloud SDK (gcloud) - Authenticated locally

✅ Python 3.12+ and your virtual environment activated

⚠️ Haven't completed Part 1? Start there first! This tutorial assumes you understand the Agent-as-Tool pattern and have ADK set up.

Important: Unlike Google Search (which can use API keys), Maps grounding REQUIRES Vertex AI. There's no fallback to the standard Gemini API.

1. Set up Environment & Install ADK

If you completed Part 1, you already have ADK installed. If starting fresh:

Create & Activate Virtual Environment:

# Create
python -m venv .venv

# Activate (each new terminal)
# macOS/Linux: source .venv/bin/activate
# Windows CMD: .venv\Scripts\activate.bat
# Windows PowerShell: .venv\Scripts\Activate.ps1

Install ADK:

pip install google-adk
pip install python-dotenv  # For environment variables

2. Enable Vertex AI and Authenticate

Enable Vertex AI API:

# Set your project ID
export PROJECT_ID="your-project-id"
gcloud config set project $PROJECT_ID

# Enable Vertex AI API
gcloud services enable aiplatform.googleapis.com

# Verify it's enabled
gcloud services list --enabled | grep aiplatform

Expected output:

aiplatform.googleapis.com    Vertex AI API

Authenticate with Application Default Credentials:

Maps grounding uses Application Default Credentials (ADC), not API keys:

# First authenticate with gcloud
gcloud auth login

# Then set up application-default credentials (this is different!)
gcloud auth application-default login

Verify authentication:

# Check your credentials file exists
ls -la ~/.config/gcloud/application_default_credentials.json

Common Issue: If you get DefaultCredentialsError, ensure you've run both gcloud auth login and gcloud auth application-default login. They serve different purposes!

3. Create Agent Project

If building on Part 1, you already have the project structure. If starting fresh:

macOS/Linux:

# Step 1: Create a new directory for your agent
mkdir maps_grounding_agent

# Step 2: Create __init__.py for the agent
echo "from . import agent" > maps_grounding_agent/__init__.py

# Step 3: Create agent.py and .env files
touch maps_grounding_agent/agent.py .env

Windows:

# Step 1: Create a new directory for your agent
mkdir maps_grounding_agent

# Step 2: Create __init__.py for the agent
echo from . import agent > maps_grounding_agent\__init__.py

# Step 3: Create agent.py and .env files
type nul > maps_grounding_agent\agent.py
type nul > .env

4. Configure Environment Variables

Open the .env file and add your configuration:

# .env

# Google API Key (optional, for non-Vertex operations)
GOOGLE_API_KEY="your-google-api-key"

# Vertex AI Configuration (REQUIRED for Maps grounding)
GOOGLE_GENAI_USE_VERTEXAI="true"
GOOGLE_CLOUD_PROJECT="your-project-id"  # e.g., "my-ai-project-123456"
GOOGLE_CLOUD_LOCATION="us-central1"      # or your preferred region

Critical Notes:

GOOGLE_GENAI_USE_VERTEXAI="true" tells ADK to route grounding tools through Vertex AI
Replace your-project-id with your actual Google Cloud Project ID
Ensure no extra quotes around values (use true not "true")

5. Edit `agent.py`

Copy and paste the following code into agent.py. This implements the Agent-as-Tool pattern, which is crucial for Maps grounding to work reliably.

maps_grounding_agent/agent.py:

"""
Google Maps Grounding Agent with ADK
Using the Agent-as-Tool pattern for reliable grounding
"""

from dotenv import load_dotenv
load_dotenv()

import os

# CRITICAL: Ensure Vertex AI environment variables are set in Python process
# This must happen BEFORE importing ADK modules
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", os.getenv("GOOGLE_GENAI_USE_VERTEXAI", "false"))
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", os.getenv("GOOGLE_CLOUD_PROJECT", ""))
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1"))

# NOW import ADK modules
from google.adk.agents import LlmAgent
from google.adk.tools import AgentTool
from google.adk.tools.google_maps_grounding_tool import GoogleMapsGroundingTool

# Helper function for Vertex AI model path
def _build_vertex_model_name(default_model: str = "gemini-2.5-flash") -> str:
    """Build fully qualified Vertex AI model name."""
    project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
    location = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")

    if project_id:
        return f"projects/{project_id}/locations/{location}/publishers/google/models/{default_model}"
    return default_model

# Create dedicated Maps grounding agent
# This agent ONLY uses GoogleMapsGroundingTool (Agent-as-Tool pattern)
maps_agent = LlmAgent(
    model=_build_vertex_model_name(),  # Use full Vertex AI path
    name='MapsAgent',
    instruction="""You are a location research specialist. Use Google Maps grounding to find 
    information about places, businesses, and locations. Always provide:
    - Full addresses
    - Ratings and review counts
    - Hours of operation when available
    - Phone numbers and websites when available
    Cite sources and be concise but comprehensive.""",
    tools=[GoogleMapsGroundingTool()],  # ONLY this tool
)

# Main agent that uses Maps grounding via AgentTool
root_agent = LlmAgent(
    name="maps_grounding_agent",
    model="gemini-2.5-flash",
    instruction="""Answer questions about places, businesses, and locations using the MapsAgent.

    When users ask about:
    - Finding restaurants, cafes, or businesses
    - Addresses and locations
    - Ratings, hours, or business details

    Always use the MapsAgent tool to get accurate, real-time information from Google Maps.
    Provide detailed, helpful responses with source attribution.""",
    description="Location-aware assistant with Google Maps grounding capabilities",
    tools=[AgentTool(agent=maps_agent)]  # Wrap Maps agent as a tool
)

Why This Pattern?

The Agent-as-Tool pattern is essential for Maps grounding. Here's why:

Isolation: The Maps agent operates in its own context, preventing conflicts with other tools
Vertex AI Routing: The full model path ensures requests go through Vertex AI, not the standard Gemini API
Environment Variables: Setting os.environ before imports ensures ADK picks up Vertex AI configuration
Tool Wrapping: Using AgentTool(agent=maps_agent) allows the main agent to delegate location queries to the specialized Maps agent

Without this pattern, you'll encounter:

ValueError: google_maps parameter is not supported in Gemini API
Inconsistent environment variable loading
Silent failures where the tool is called but produces no results

Now you would have the following directory structure:

my_project/
    maps_grounding_agent/
        __init__.py
        agent.py
    .env

6. Run Your Agent

There are multiple ways to interact with your agent:

Dev UI (adk web):

adk web

Note for Windows users: If you hit _make_subprocess_transport NotImplementedError, use adk web --no-reload instead.

Step 1: Open the URL provided (usually http://localhost:8000 or http://127.0.0.1:8000) in your browser.

Step 2: In the top-left corner, select "maps_grounding_agent" from the dropdown.

Step 3: Start asking location-based questions!

Terminal (adk run):

adk run maps_grounding_agent

To exit, use Cmd/Ctrl+C.

📝 Example Prompts to Try

With these questions, you can confirm that the agent is actually calling Google Maps grounding:

"Find 3 highly rated restaurants in Times Square, New York"
"What is the address of the Empire State Building?"
"Find coffee shops near Central Park"
"What are the hours for Starbucks in downtown Seattle?"
"Tell me about the Louvre Museum in Paris"

You should see the agent return specific place information with addresses, ratings, hours, and other details from Google Maps.

Example Response:

Here are 3 highly rated restaurants in Times Square, New York:

1. **Carmine's Italian Restaurant**
   📍 Address: 200 W 44th St, New York, NY 10036
   ⭐ Rating: 4.5/5 (8,234 reviews)
   🕒 Hours: 11:30 AM - 11:00 PM
   Known for family-style Italian cuisine

2. **Ellen's Stardust Diner**
   📍 Address: 1650 Broadway, New York, NY 10019
   ⭐ Rating: 4.3/5 (12,456 reviews)
   🕒 Hours: 7:00 AM - 12:00 AM
   Famous for singing waitstaff

3. **Junior's Restaurant**
   📍 Address: 1515 Broadway, New York, NY 10036
   ⭐ Rating: 4.4/5 (6,789 reviews)
   🕒 Hours: 6:30 AM - 12:00 AM
   Iconic for cheesecake

🎉 You've successfully created and interacted with your Google Maps grounding agent using ADK!

How Grounding with Google Maps Works

Grounding with Google Maps is the process that connects your agent to Google's database of 200+ million places worldwide, allowing it to generate accurate responses based on real-world location data. When a user's prompt requires location-based information, the agent's underlying LLM intelligently decides to invoke the GoogleMapsGroundingTool to find relevant place information.

Data Flow Diagram

This diagram illustrates the step-by-step process of how a user query results in a grounded response:

Detailed Description

The Maps grounding agent uses the data flow described in the diagram to retrieve, process, and incorporate location information into the final answer presented to the user.

User Query: An end-user interacts with your agent by asking a location-based question (e.g., "Find restaurants near Times Square").
ADK Orchestration: The Agent Development Kit orchestrates the agent's behavior and passes the user's message to your agent.
LLM Analysis and Tool-Calling: The agent's LLM (e.g., a Gemini model) analyzes the prompt. If it determines that location information is required, it triggers the grounding mechanism by calling the Maps agent through AgentTool. This is ideal for answering queries about restaurants, businesses, addresses, or any location-based information.
Google Maps Service Interaction: The GoogleMapsGroundingTool interacts with Google Maps' database through Vertex AI (not directly through the standard Gemini API). The service formulates and executes search queries against Google's place database.
Place Retrieval & Ranking: Google Maps retrieves and ranks the most relevant places based on the query, location context, and relevance scoring. This includes addresses, ratings, reviews, hours, and other place attributes.
Context Injection: The Maps service integrates the retrieved place information into the model's context before the final response is generated. This crucial step allows the model to "reason" over real-world location data.
Grounded Response Generation: The LLM, now informed by relevant Google Maps data, generates a response that incorporates the retrieved place information with proper formatting and context.
Response Presentation with Sources: The ADK receives the final grounded response, which includes place references and grounding metadata, and presents it to the user with attribution. This allows end-users to verify the information against Google Maps.

Understanding Grounding with Google Maps Response

When the agent uses Google Maps grounding, it returns detailed information that includes the final text answer and metadata about the places used to generate that answer. This metadata is crucial for verifying the response and providing attribution.

Example of a Grounded Response

The following is an example of the content returned by the model after a grounded query against Google Maps.

Final Answer Text:

Here are 3 highly rated restaurants in Times Square, New York:

1. **Carmine's Italian Restaurant**
   Address: 200 W 44th St, New York, NY 10036
   Rating: 4.5/5 stars (8,234 reviews)
   Hours: 11:30 AM - 11:00 PM
   Known for family-style Italian cuisine

2. **Ellen's Stardust Diner**
   Address: 1650 Broadway, New York, NY 10019
   Rating: 4.3/5 stars (12,456 reviews)
   Hours: 7:00 AM - 12:00 AM
   Famous for singing waitstaff

3. **Junior's Restaurant**
   Address: 1515 Broadway, New York, NY 10036
   Rating: 4.4/5 stars (6,789 reviews)
   Hours: 6:30 AM - 12:00 AM
   Iconic for cheesecake and New York-style deli

Grounding Metadata Snippet:

This is the grounding metadata you will receive when inspecting events or using adk web (on the Response tab):

{
  "groundingMetadata": {
    "groundingChunks": [
      {
        "googleMaps": {
          "title": "Carmine's Italian Restaurant",
          "uri": "https://maps.google.com/maps?cid=1234567890",
          "placeId": "ChIJ...",
          "address": "200 W 44th St, New York, NY 10036"
        }
      },
      {
        "googleMaps": {
          "title": "Ellen's Stardust Diner",
          "uri": "https://maps.google.com/maps?cid=9876543210",
          "placeId": "ChIJ...",
          "address": "1650 Broadway, New York, NY 10019"
        }
      },
      {
        "googleMaps": {
          "title": "Junior's Restaurant",
          "uri": "https://maps.google.com/maps?cid=1122334455",
          "placeId": "ChIJ...",
          "address": "1515 Broadway, New York, NY 10036"
        }
      }
    ],
    "groundingSupports": [
      {
        "groundingChunkIndices": [0, 1, 2],
        "segment": {
          "endIndex": 450,
          "startIndex": 0,
          "text": "Here are 3 highly rated restaurants in Times Square, New York..."
        }
      }
    ],
    "retrievalQueries": [
      "highly rated restaurants Times Square New York",
      "best restaurants near Times Square"
    ]
  }
}

How to Interpret the Response

The metadata provides a link between the text generated by the model and the Google Maps places that support it. Here is a step-by-step breakdown:

groundingChunks: This is a list of the Google Maps places the model consulted. Each chunk contains:
- googleMaps: Object with place details
- title: Name of the place
- uri: Google Maps URL for the place
- placeId: Unique Google Maps Place ID
- address: Full address of the place
groundingSupports: This list connects specific sentences in the final answer back to the groundingChunks.
- segment: This object identifies a specific portion of the final text answer, defined by its startIndex, endIndex, and the text itself.
- groundingChunkIndices: This array contains the index numbers that correspond to the sources listed in the groundingChunks. For example, [0, 1, 2] means this text segment is supported by all three places.
retrievalQueries: This array shows the specific search queries that were executed against Google Maps to find relevant places.

How to Display Grounding Responses with Google Maps

Unlike Google Search grounding, Maps grounding does not require specific display components. However, displaying place information with proper formatting enhances user experience and builds trust.

Optional Citation Display

Since grounding metadata is provided, you can choose to implement place displays based on your application needs:

Simple Text Display (Minimal Implementation):

for event in events:
    if event.is_final_response():
        print(event.content.parts[0].text)

        # Optional: Show source count
        if event.grounding_metadata:
            chunks = event.grounding_metadata.grounding_chunks
            print(f"\nBased on {len(chunks)} places from Google Maps")

Enhanced Place Display (Recommended):

You can implement detailed place displays that show addresses, ratings, and links:

def display_maps_response(event):
    """Display Maps grounding response with enhanced place information"""

    # Show the main response
    if event.content and event.content.parts:
        print(event.content.parts[0].text)

    # Show place details if available
    if hasattr(event, 'grounding_metadata') and event.grounding_metadata:
        metadata = event.grounding_metadata

        if hasattr(metadata, 'grounding_chunks') and metadata.grounding_chunks:
            print(f"\n Sources ({len(metadata.grounding_chunks)} places):")

            for i, chunk in enumerate(metadata.grounding_chunks, 1):
                if hasattr(chunk, 'google_maps'):
                    place = chunk.google_maps
                    print(f"\n{i}. {place.title}")

                    if hasattr(place, 'address'):
                        print(f"   {place.address}")

                    if hasattr(place, 'uri'):
                        print(f"   View on Maps: {place.uri}")

                    if hasattr(place, 'place_id'):
                        print(f"   Place ID: {place.place_id}")

Implementation Considerations

When implementing Google Maps grounding displays:

Place Attribution: Always show that information comes from Google Maps
Simple Integration: Basic text output requires no additional display logic
Optional Enhancements: Add formatted displays with emojis, links, and structured data
Map Links: The uri field provides direct Google Maps links for each place
Place IDs: The placeId field can be used for further integration with Google Maps APIs
Search Queries: The retrievalQueries array shows what searches were performed

Current Limitations and Workarounds

While Google Maps Grounding is powerful, it's important to understand its current state and limitations compared to other grounding tools like Vertex AI Search.

What Maps Grounding CAN Do

✅ Find businesses, restaurants, and points of interest

✅ Provide addresses, ratings, and basic place information

✅ Return general location details and descriptions

✅ Ground responses with Google Maps data

✅ Work through Vertex AI with proper authentication

What Maps Grounding CANNOT Do (Yet)

❌ Direct Places API Integration: Unlike direct Places API calls, Maps grounding doesn't provide:

Detailed place photos
Full review text and user-generated content
Advanced place details (wheelchair accessibility, parking info, etc.)
Real-time "currently open" status

❌ Advanced Filtering: Limited ability to filter by:

Price range ($ - $$$$)
Specific amenities (outdoor seating, reservations, etc.)
Distance radius from a specific point

❌ Directions Integration: Cannot provide:

Turn-by-turn directions
Travel time estimates
Route alternatives

❌ Custom Maps Display: No built-in support for:

Interactive map widgets
Custom map markers
Map visualizations

Why the Agent-as-Tool Pattern is Required

Based on hours of debugging and reviewing the ADK source code, here's why the Agent-as-Tool pattern is essential:

The Problem:

# ❌ This approach doesn't work reliably!
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.tools.google_maps_grounding_tool import GoogleMapsGroundingTool

agent = LlmAgent(
    model='gemini-2.5-flash',
    tools=[
        GoogleMapsGroundingTool(),           # Built-in grounding tool
        FunctionTool(my_custom_function),    # Custom function
    ]
)

Why it fails:

Environment variables aren't consistently picked up by the Python process
The standard model name doesn't route through Vertex AI
Tool execution context conflicts with other FunctionTools
Results in: ValueError: google_maps parameter is not supported in Gemini API

The Solution (Agent-as-Tool Pattern):

# ✅ This works!

# Step 1: Set environment variables in Python process
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "true")
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", "your-project-id")
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "us-central1")

# Step 2: Create dedicated Maps agent with Vertex model path
maps_agent = LlmAgent(
    model=f"projects/{project}/locations/{location}/publishers/google/models/gemini-2.5-flash",
    name='MapsAgent',
    tools=[GoogleMapsGroundingTool()],  # ONLY this tool
)

# Step 3: Wrap as a tool for the main agent
main_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='MainAgent',
    tools=[
        AgentTool(agent=maps_agent),        # Wrapped Maps agent
        FunctionTool(my_custom_function),   # Your custom tools
    ]
)

Why this works:

Environment Variables: os.environ.setdefault() ensures variables are in the process environment before ADK imports
Vertex AI Routing: Full model path forces requests through Vertex AI
Context Isolation: The Maps agent operates independently, preventing tool conflicts
Proper Delegation: The main agent delegates location queries to the specialized Maps agent

This pattern is recommended by the ADK team for all built-in grounding tools and is documented in the ADK built-in tools guide.

Workarounds We Developed

For advanced use cases requiring Places API features, consider hybrid approaches:

def enhanced_place_lookup(place_id: str):
    """
    Hybrid approach: Combine Maps grounding with Places API

    1. Use Maps grounding for initial search
    2. Extract place_id from grounding metadata
    3. Use Places API for detailed information
    """

    # Step 1: Maps grounding gives basic info + place_id
    maps_result = get_place_from_grounding(query)

    # Step 2: Use place_id with Places API for details
    if maps_result and maps_result.place_id:
        detailed_info = places_api.get_place_details(maps_result.place_id)

        # Combine both sources
        return {
            "basic_info": maps_result,    # From grounding
            "detailed_info": detailed_info # From Places API
        }

    return maps_result

Note: This hybrid approach requires:

Additional Places API setup and billing
Separate API key or credentials
Compliance with Places API terms of service

For most use cases, Maps grounding alone is sufficient and provides accurate, up-to-date location information.

Troubleshooting Common Issues

Issue 1: `ValueError: google_maps parameter is not supported`

Symptoms:

ValueError: google_maps parameter is not supported in Gemini API.
Use Vertex AI for grounding with Google Maps.

Causes:

GOOGLE_GENAI_USE_VERTEXAI not set to "true"
Environment variables not loaded into Python process
Not using the full Vertex AI model path

Solution:

# At the TOP of agent.py (before ADK imports)
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "true")
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", "your-project-id")
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "us-central1")

# Use full Vertex model path
model = f"projects/{project}/locations/{location}/publishers/google/models/gemini-2.5-flash"

Issue 2: `DefaultCredentialsError`

Symptoms:

google.auth.exceptions.DefaultCredentialsError: Could not automatically 
determine credentials.

Cause: Application Default Credentials not set up

Solution:

# Run BOTH commands:
gcloud auth login                      # Step 1: Authenticate gcloud
gcloud auth application-default login  # Step 2: Set up ADC (different!)

# Verify credentials exist
ls ~/.config/gcloud/application_default_credentials.json

Issue 3: Agent Doesn't Use Maps Grounding

Symptoms: Agent responds but doesn't call Maps grounding tool

Causes:

Query not specific enough
Agent instruction doesn't indicate when to use Maps
Location doesn't exist or is too vague

Solution:

# Be explicit in agent instructions
instruction="""
IMPORTANT: For ALL location-based queries, use the MapsAgent tool.

Location-based queries include:
- "Find [business type] in [location]"
- "What is the address of [place]?"
- "Tell me about [landmark]"
- "Where is [business name]?"
- Any query mentioning cities, neighborhoods, or specific places

ALWAYS delegate these queries to MapsAgent!
"""

# Use specific queries
"Find restaurants in Times Square, New York"  # ✅ Good - specific location
"Find good food"                               # ❌ Bad - too vague

Issue 4: Slow Response Times

Symptoms: Maps queries take 5-10 seconds

Explanation: This is normal! Maps grounding involves:

Geocoding the location (1-2 seconds)
Searching Google Maps database (2-3 seconds)
Ranking and filtering results (1 second)
LLM synthesis (1-2 seconds)

Total: 5-8 seconds is expected

Optimization tips:

# Use streaming to show progress
async for event in runner.run_async(query, session_id):
    if event.content:
        for part in event.content.parts:
            if part.text:
                print(part.text, end='', flush=True)  # Stream as it arrives

Issue 5: Missing Place Details

Symptoms: Results missing ratings, hours, or other details

Cause: Not all places have complete data in Google Maps

Solution:

# Update instruction to handle missing data gracefully
instruction="""
When presenting Maps results:
1. Always include what data IS available
2. If ratings are missing, state "Rating not available"
3. If hours are missing, suggest "Check Google Maps for current hours"
4. Never hallucinate or make up missing information
5. Always cite Google Maps as the source
"""

Summary

Google Maps Grounding transforms AI agents from general-purpose assistants into location-aware systems capable of providing accurate, place-attributed information from Google Maps' vast database. By integrating this feature into your ADK agents, you enable them to:

Access place information from Google Maps' database of 200+ million places worldwide
Provide location attribution for transparency and trust in place recommendations
Deliver comprehensive answers with verifiable place facts including addresses, ratings, and hours
Maintain real-time accuracy within Google's constantly updated place database

The grounding process seamlessly connects user queries to Google Maps' place database, enriching responses with relevant location context while maintaining conversational flow. With proper implementation using the Agent-as-Tool pattern, your agents become powerful tools for location-based information discovery and decision-making.

Key Implementation Requirements

Vertex AI is Mandatory: No fallback to standard Gemini API
Agent-as-Tool Pattern: Essential for reliable grounding
Environment Variables: Must be set in Python process before ADK imports
Full Model Path: Use projects/{project}/locations/{location}/publishers/google/models/{model}
Application Default Credentials: ADC authentication required

Current State and Future Expectations

What Works Today:

✅ Place search and recommendations
✅ Addresses, ratings, and basic details
✅ Integration with ADK's Agent-as-Tool pattern
✅ Reliable grounding through Vertex AI

Known Limitations:

❌ No direct Places API integration for advanced details
❌ Limited real-time availability checking
❌ No custom map rendering capabilities
❌ No turn-by-turn directions integration

What's Next:
As Google continues to develop the ADK platform, we expect:

Enhanced Places API integration
More comprehensive place details
Real-time place availability
Better documentation and examples

Until then, this article serves as the de facto documentation for Maps grounding with ADK.

What's Next?

🎉 Congratulations! You now have a working Google Maps grounding agent!

Coming in Part 3: Full-Stack Frontend

We'll build a beautiful UI for your agent with:

CopilotKit for the chat interface
AG-UI Protocol for real-time agent communication
Next.js frontend with modern UI
Streaming responses for better UX

The Complete 5-Part Series

✅ Part 1: Google Search Grounding with ADK
✅ Part 2: Google Maps Grounding with ADK (completed!)
Part 3: Full-Stack Frontend with CopilotKit & AG-UI
Part 4: Persistent Sessions with PostgreSQL & Docker
Part 5: Production Deployment on Cloud Run

By Part 5, you'll have a production-ready, scalable AI agent deployed to the cloud with persistent sessions and a beautiful UI!

Resources

📦 Full Code: https://github.com/Greyisheep/ag-ui-adk-grounding-app
🗺️ Google Maps Grounding Docs: https://cloud.google.com/vertex-ai/generative-ai/docs/grounding/grounding-with-google-maps
🔍 ADK Source Code: https://github.com/google/adk-python/blob/main/src/google/adk/tools/google_maps_grounding_tool.py
📚 ADK Documentation: https://google.github.io/adk-docs/
📖 Vertex AI Search Grounding (for comparison): https://google.github.io/adk-docs/grounding/vertex_ai_search_grounding/

Questions or Issues?

If you're struggling with Maps grounding, drop a comment! I spent hours debugging this so you don't have to. This article documents everything that's not in the official docs yet.

⭐ Star the repo if this saved you time, and follow me for Part 3 where we build the frontend!

About the Author: I'm Claret, an AI Engineer building production agentic systems. When Google's ADK documentation didn't exist for Maps grounding, I dug through source code and debugged for hours to figure it out. This series shares what I learned so you can skip the struggle and use this as your Maps grounding documentation.

Connect with me on LinkedIn | GitHub

Building AI Agents with Google Search Grounding and ADK (Part 1/5)

Claret Ibeawuchi — Thu, 02 Oct 2025 08:40:12 +0000

This is Part 1 of a 5-part series on building production-ready AI agents with Google's Agent Development Kit (ADK):

Part 1: Google Search Grounding with ADK (you are here)

Part 2: Adding Google Maps Grounding with Vertex AI

Part 3: Building a Full-Stack Frontend with CopilotKit & AG-UI

Part 4: Persistent Sessions with PostgreSQL & Docker

Part 5: Production Deployment on Cloud Run

Why Search Grounding Matters

Most AI models are trained on data with a cutoff date. About a year ago, if you asked ChatGPT about events from last week, and it can't help you. Or Claude about today's stock prices, and it will apologise. This is the knowledge cutoff problem, and it's a major limitation for production AI applications.

Search Grounding - the ability to ground your AI agent's responses in real-time web data. Instead of hallucinating or saying "I don't know," your agent can:

✅ Search the web for current information

✅ Cite sources with clickable links

✅ Provide factual answers backed by real data

✅ Stay updated without retraining the model

According to Google Cloud's documentation, grounding ensures responses are based on the latest and most accurate information, which is critical for:

News and current events - "What's happening in AI this week?"
Financial data - "What's Tesla's stock price today?"
Technical troubleshooting - "How do I fix the latest React 19 breaking changes?"
Research - "What are the recent findings on climate change?"

The Challenge

Implementing search grounding isn't straightforward. When I built my first search-grounded agent, I spent 3 hours debugging the cryptic error: "Function search is not found".

The issue? Google's Search grounding tool can't be mixed directly with other function tools. You need a specific pattern — the Agent-as-Tool pattern — which we'll implement in this tutorial.

What we'll build: A Python-based AI agent using ADK that can search the web in real-time and provide current, source-attributed answers.

🔗 Full Code: https://github.com/Greyisheep/ag-ui-adk-grounding-app

Prerequisites

Before we start, make sure you have:

Python 3.12+ installed
A Google AI Studio API key (Get one here)
Basic knowledge of Python and async programming
15-20 minutes of your time

Understanding Google Search Grounding

What is Grounding?

Grounding is the process of connecting an LLM's responses to external, verifiable sources. Instead of relying solely on training data, the model:

Executes a search query based on the user's question
Retrieves relevant web results from Google Search
Processes the search results to extract information
Generates a response grounded in those results
Returns source attribution with links and metadata

The Benefits

According to Google's grounding documentation, you can:

✅ Ensure responses are based on current, accurate information
✅ Retrieve artifacts from the web for analysis and reasoning
✅ Find region-specific information with localization
✅ Perform technical troubleshooting with latest docs
✅ Provide source attribution for transparency and trust

Real-World Use Cases

1. Customer Support Agents

User: "How do I fix the error in the latest version of your API?"
Agent: *Searches documentation* → Provides fix with link to release notes

2. Research Assistants

User: "What are the latest developments in quantum computing?"
Agent: *Searches recent papers* → Summarizes findings with citations

3. News Aggregation

User: "What's the top tech news today?"
Agent: *Searches news sites* → Delivers current stories with sources

4. Financial Analysis

User: "What's driving the market today?"
Agent: *Searches financial news* → Provides analysis with data sources

Part 1: Setting Up Your ADK Project

Let's create a Python project with ADK from scratch.

1.1 Create Project Structure

# Create project directory
mkdir adk-search-agent
cd adk-search-agent

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Create project files
touch agent.py
touch requirements.txt
touch .env
touch .gitignore

1.2 Add Dependencies

Create requirements.txt:

google-adk
google-genai
python-dotenv

Install dependencies:

pip install -r requirements.txt

1.3 Configure Environment Variables

Create .env file:

# .env
GOOGLE_API_KEY="your-google-api-key-here"

Create .gitignore:

# .gitignore
venv/
.env
*.log
__pycache__/
*.pyc

Security Note: Never commit your .env file! The .gitignore ensures your API key stays secure.

Part 2: The Agent-as-Tool Pattern Explained

This is the critical pattern you need to understand for search grounding.

The Problem: Direct Tool Mixing Fails

My first attempt looked like this:

# This approach doesn't work!
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.tools.google_search_tool import GoogleSearchTool

def my_custom_function() -> str:
    """Custom business logic"""
    return "Some result"

# This will fail!
agent = LlmAgent(
    model='gemini-2.5-flash',
    tools=[
        GoogleSearchTool(),                    # Built-in grounding tool
        FunctionTool(my_custom_function),      # Custom function
    ]
)

Error: "Function search is not found"

Why it fails:

Google Search grounding operates in its own execution context
It uses Google's Search API with special handling
Mixing it directly with FunctionTools causes context conflicts
The ADK can't properly resolve the search function calls

The Solution: Agent-as-Tool Pattern

The fix is to isolate the search tool in a dedicated agent, then wrap that agent as a tool:

# This works!
from google.adk.tools import AgentTool

# Step 1: Create a dedicated search agent
search_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='SearchAgent',
    instruction="You are a specialist in Google Search grounding.",
    tools=[GoogleSearchTool()],  # ONLY the search tool
)

# Step 2: Wrap it as a tool for the main agent
main_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='MainAgent',
    instruction="You are a helpful assistant.",
    tools=[
        AgentTool(agent=search_agent),        # Wrapped search agent
        FunctionTool(my_custom_function),     # Your custom tools
    ]
)

Why This Pattern Works

Isolation: Each agent has its own context

The search_agent focuses solely on search grounding
No conflicts with other tools or execution contexts

Delegation: The main agent delegates to the search agent

When the main agent needs current information, it calls the search agent
The search agent executes the search and returns results
The main agent incorporates those results into its response

Flexibility: You can add multiple specialized agents

Search agent for web grounding
Maps agent for location grounding (Part 2!)
Database agent for data retrieval
Each isolated in its own context

Recommended by Google: This pattern is documented in the ADK built-in tools guide as the correct approach for grounding tools.

Part 3: Building Your Search-Grounded Agent

Now let's build the complete agent. We'll start simple and add complexity.

3.1 Basic Setup and Imports

Create agent.py and add:

"""
ADK Search-Grounded Agent
Demonstrates the Agent-as-Tool pattern for Google Search grounding
"""

from dotenv import load_dotenv
load_dotenv()

import os
import asyncio
import logging
from datetime import datetime

# ADK imports
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool, AgentTool
from google.adk.tools.google_search_tool import GoogleSearchTool
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Verify API key is loaded
if not os.getenv('GOOGLE_API_KEY'):
    raise ValueError("GOOGLE_API_KEY not found in environment variables")

logger.info("ADK Search Agent initialized")

3.2 Add a Custom Tool (Optional)

Let's add a simple custom function to demonstrate tool mixing:

def get_current_time() -> str:
    """
    Returns the current date and time.
    Useful for time-sensitive queries.
    """
    current = datetime.now()
    return f"Current time: {current.strftime('%Y-%m-%d %H:%M:%S %Z')}"

logger.info("Custom tools defined")

3.3 Create the Search Agent

This agent is dedicated to search grounding only:

# Create dedicated search agent
search_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='SearchAgent',
    instruction="""You are a specialist in Google Search grounding.

    Your role:
    - Use web search to find current, factual information
    - Provide structured, well-organized findings
    - Always cite your sources with URLs
    - Focus on accuracy and recency

    When you find information, format it clearly with:
    1. Direct answer to the query
    2. Supporting details and context
    3. Source attribution with links""",
    tools=[GoogleSearchTool()],  # ONLY this tool - critical!
)

logger.info("Search agent created")

Key points:

model='gemini-2.5-flash': Fast, cost-effective for search tasks
name='SearchAgent': Helpful for debugging and logging
instruction: Clear, specific role definition
tools=[GoogleSearchTool()]: Only the search tool — no mixing!

3.4 Create the Main Agent

The main agent coordinates everything:

# Create main agent with search capability
main_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='MainAgent',
    instruction="""You are a helpful AI assistant with access to:

    1. Real-time web search for current information
    2. Current time/date information

    Guidelines:
    - When asked about CURRENT events, RECENT news, or TIME-SENSITIVE 
      information, you MUST use the search capability
    - When asked about the time or date, use the time tool
    - Always cite sources when using search results
    - Be concise but comprehensive
    - Admit when you don't have current information and suggest using search

    Example queries that require search:
    - "What's the latest news about [topic]?"
    - "What happened today in [domain]?"
    - "Recent developments in [field]"
    - "Current status of [event]"
    """,
    tools=[
        AgentTool(agent=search_agent),      # Search grounding
        FunctionTool(get_current_time),     # Custom function
    ]
)

logger.info("Main agent created with search capability")

3.5 Setup Runner and Session Management

# Create session service (in-memory for now, PostgreSQL in Part 4!)
session_service = InMemorySessionService()

# Create runner
runner = Runner(
    agent=main_agent,
    session_service=session_service,
    app_name="search_agent_app"
)

logger.info("Runner initialized")

Part 4: Testing Your Search Agent

4.1 Create a Query Function

Add this helper function to run queries:

async def run_query(query: str, session_id: str = None):
    """
    Run a single query through the agent

    Args:
        query: The user's question
        session_id: Optional session ID for conversation continuity
    """
    print(f"\n{'='*60}")
    print(f"🔍 QUERY: {query}")
    print(f"{'='*60}\n")

    # Create or reuse session
    if not session_id:
        session = session_service.create_session_sync()
        session_id = session.session_id

    # Run the agent
    response_text = ""
    async for event in runner.run_async(
        user_message=query,
        session_id=session_id
    ):
        # Collect text content from events
        if hasattr(event, 'content') and event.content:
            for part in event.content.parts:
                if hasattr(part, 'text') and part.text:
                    response_text += part.text

    print(f"RESPONSE:")
    print(f"{response_text}\n")
    print(f"{'-'*60}\n")

    return session_id

4.2 Create Test Cases

Add a main function with test queries:

async def main():
    """Run test queries"""

    print("\n" + "="*60)
    print("ADK Search-Grounded Agent - Test Suite")
    print("="*60 + "\n")

    # Test 1: Current events (requires search)
    await run_query(
        "What are the latest developments in artificial intelligence this week?"
    )

    # Test 2: Recent news (requires search)
    await run_query(
        "What are today's top technology news stories?"
    )

    # Test 3: Factual query with recency (requires search)
    await run_query(
        "Tell me about recent breakthroughs in quantum computing"
    )

    # Test 4: Simple time query (uses custom function)
    await run_query(
        "What's the current time?"
    )

    # Test 5: Mixed query (might use both)
    await run_query(
        "What major tech events happened today?"
    )

    print("\n" + "="*60)
    print("Test suite completed!")
    print("="*60 + "\n")

if __name__ == "__main__":
    asyncio.run(main())

4.3 Run Your Agent

python agent.py

4.4 Expected Output

You should see output like this:

============================================================
ADK Search-Grounded Agent - Test Suite
============================================================

============================================================
QUERY: What are the latest developments in artificial intelligence this week?
============================================================

RESPONSE:
Based on recent web sources, here are the latest AI developments:

1. **OpenAI's GPT-4 Vision Capabilities**
   OpenAI has announced enhanced multimodal capabilities for GPT-4, 
   allowing it to process and analyze images alongside text.
   Source: https://techcrunch.com/2024/...

2. **EU AI Act Implementation**
   The European Union has begun enforcing its comprehensive AI 
   regulations, affecting companies worldwide.
   Source: https://reuters.com/technology/...

3. **Google's Gemini Advanced**
   Google released Gemini Advanced with improved reasoning and 
   coding capabilities.
   Source: https://blog.google/technology/...

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

Part 5: Understanding Grounding Metadata

One of the most powerful features of search grounding is the metadata returned with each response.

5.1 Metadata Structure

According to Google's documentation, the response includes:

{
  "content": "The AI-generated response text",
  "groundingMetadata": {
    "webSearchQueries": ["actual search query executed"],
    "groundingChunks": [
      {
        "web": {
          "uri": "https://example.com/article",
          "title": "Article Title"
        }
      }
    ],
    "groundingSupports": [
      {
        "segment": {
          "text": "Specific text segment from response",
          "startIndex": 0,
          "endIndex": 50
        },
        "groundingChunkIndices": [0],
        "confidenceScores": [0.95]
      }
    ]
  }
}

Key Fields:

webSearchQueries: The actual search queries the agent executed
groundingChunks: Web sources used, with URLs and titles
groundingSupports: Maps response segments to sources
confidenceScores: Confidence in the grounding (0.0 to 1.0)

5.2 Accessing Metadata in Code

Enhance your run_query function to display metadata:

async def run_query_with_metadata(query: str):
    """Run query and display grounding metadata"""

    print(f"\n🔍 Query: {query}\n")

    session = session_service.create_session_sync()

    sources = []
    response_text = ""

    async for event in runner.run_async(
        user_message=query,
        session_id=session.session_id
    ):
        # Collect response text
        if hasattr(event, 'content') and event.content:
            for part in event.content.parts:
                if hasattr(part, 'text') and part.text:
                    response_text += part.text

        # Collect grounding metadata
        if hasattr(event, 'grounding_metadata'):
            metadata = event.grounding_metadata

            if hasattr(metadata, 'web_search_queries'):
                print(f"Search Queries Executed:")
                for query in metadata.web_search_queries:
                    print(f"   - {query}")

            if hasattr(metadata, 'grounding_chunks'):
                print(f"\nSources ({len(metadata.grounding_chunks)}):")
                for i, chunk in enumerate(metadata.grounding_chunks[:5], 1):
                    if hasattr(chunk, 'web'):
                        print(f"   {i}. {chunk.web.title}")
                        print(f"      {chunk.web.uri}")
                        sources.append(chunk.web.uri)

            if hasattr(metadata, 'grounding_supports'):
                print(f"\nGrounding Confidence:")
                for support in metadata.grounding_supports[:3]:
                    if hasattr(support, 'confidence_scores'):
                        avg_confidence = sum(support.confidence_scores) / len(support.confidence_scores)
                        print(f"   - {avg_confidence:.2%}")

    print(f"\nResponse:\n{response_text}\n")

    return sources

Part 6: The Complete Execution Flow

Understanding what happens under the hood:

1. User asks: "What's the latest AI news?"
        ↓
2. Main Agent receives query
        ↓
3. Main Agent analyzes: "This requires current information"
        ↓
4. Main Agent decides to call AgentTool(search_agent)
        ↓
5. Search Agent receives the delegated query
        ↓
6. Search Agent calls GoogleSearchTool()
        ↓
7. Google Search API:
   - Executes web search
   - Retrieves relevant results
   - Extracts content and metadata
        ↓
8. Search results return to Search Agent
        ↓
9. Search Agent processes and structures the information
        ↓
10. Structured results return to Main Agent
        ↓
11. Main Agent synthesizes final response
        ↓
12. Response delivered to user with sources

Why Separate Agents?

Execution Context Isolation

Search grounding needs its own event loop
Prevents conflicts with other async operations
Allows proper error handling and retries

Clear Separation of Concerns

Search Agent: "I only do search, and I do it well"
Main Agent: "I coordinate and synthesize"
Each agent has a focused, well-defined role

Scalability

Easy to add more specialized agents (Maps in Part 2!)
Can run agents in parallel if needed
Better resource management

Debugging

Errors are isolated to specific agents
Easier to log and trace execution
Clear boundaries for testing

Part 7: Common Issues & Solutions

Issue 1: "Function search is not found"

Symptoms:

Error: Function 'search' is not found in the available tools

Cause: Mixing GoogleSearchTool() directly with other FunctionTools

Solution: Use the Agent-as-Tool pattern

# Correct
search_agent = LlmAgent(tools=[GoogleSearchTool()])
main_agent = LlmAgent(tools=[AgentTool(agent=search_agent)])

# Incorrect
agent = LlmAgent(tools=[GoogleSearchTool(), FunctionTool(my_func)])

Issue 2: Agent Doesn't Search When It Should

Symptoms: Agent gives generic responses without using search

Cause: Instruction doesn't clearly indicate when to search

Solution: Be explicit and use imperative language

instruction="""
When asked about CURRENT events, RECENT news, or TIME-SENSITIVE 
information, you MUST use the search capability.

Keywords that require search:
- "latest", "recent", "today", "this week"
- "current", "now", "happening"
- Any specific dates or timeframes
"""

Issue 3: API Key Not Found

Symptoms: ValueError: GOOGLE_API_KEY not found

Solution:

# Check .env file exists and has correct format
cat .env

# Verify it loads correctly
python -c "from dotenv import load_dotenv; load_dotenv(); import os; print(os.getenv('GOOGLE_API_KEY'))"

# Make sure no extra quotes or spaces
# ✅ Correct: GOOGLE_API_KEY=AIzaSy...
# ❌ Wrong: GOOGLE_API_KEY="AIzaSy..."  (remove quotes)

Issue 4: ModuleNotFoundError

Symptoms: ModuleNotFoundError: No module named 'google.adk'

Solution:

# Ensure virtual environment is activated
which python  # Should show path to venv/bin/python

# If not activated:
source venv/bin/activate

# Reinstall dependencies
pip install --upgrade pip
pip install -r requirements.txt

# Verify installation
pip show google-adk

Issue 5: Slow Search Responses

Symptoms: Queries take 5-10 seconds to respond

Explanation: This is normal! Search grounding involves:

Executing web searches (1-2 seconds)
Retrieving and processing results (1-2 seconds)
LLM synthesis (1-2 seconds)

Optimization tips (for later):

Cache common queries
Use streaming responses to show progress
Set timeout limits for searches
Implement async patterns for better UX

Issue 6: Empty Search Results

Symptoms: Agent says "I couldn't find information"

Causes:

Query is too vague
Search terms are too specific/obscure
Network issues

Solution:

# Add fallback instructions
instruction="""
...
If search returns no results:
1. Try rephrasing the search query
2. Use broader search terms
3. Inform the user politely and suggest alternative queries
"""

Part 8: Production-Ready Enhancements

8.1 Add Comprehensive Logging

import logging

# Configure detailed logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("agent.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Add callbacks for visibility
def log_before_agent(ctx):
    logger.info(f"Agent called: {ctx.agent.name}")
    if hasattr(ctx, 'user_message'):
        logger.info(f"Query: {ctx.user_message}")

def log_after_agent(ctx):
    logger.info(f"Agent completed: {ctx.agent.name}")

# Apply to agents
search_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='SearchAgent',
    tools=[GoogleSearchTool()],
    before_agent_callback=log_before_agent,
    after_agent_callback=log_after_agent,
)

View logs:

tail -f agent.log

8.2 Add Error Handling

async def run_query_safely(query: str):
    """Run query with error handling"""
    try:
        session = session_service.create_session_sync()
        response_text = ""

        async for event in runner.run_async(
            user_message=query,
            session_id=session.session_id
        ):
            if hasattr(event, 'content') and event.content:
                for part in event.content.parts:
                    if hasattr(part, 'text'):
                        response_text += part.text

        return response_text

    except Exception as e:
        logger.error(f"Error running query: {e}")
        return f"Sorry, I encountered an error: {str(e)}"

8.3 Add Rate Limiting (For Production)

from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_requests=10, time_window=60):
        self.max_requests = max_requests
        self.time_window = time_window  # seconds
        self.requests = []

    def can_proceed(self):
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.time_window)

        # Remove old requests
        self.requests = [r for r in self.requests if r > cutoff]

        # Check limit
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True
        return False

# Usage
limiter = RateLimiter(max_requests=10, time_window=60)

async def run_query_with_limit(query: str):
    if not limiter.can_proceed():
        return "Rate limit exceeded. Please wait a moment."

    return await run_query(query)

Complete Working Code

Here's the full agent.py with all enhancements:

"""
ADK Search-Grounded Agent - Complete Implementation
Part 1/5 of the ADK Production Agent Series
"""

from dotenv import load_dotenv
load_dotenv()

import os
import asyncio
import logging
from datetime import datetime

# ADK imports
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool, AgentTool
from google.adk.tools.google_search_tool import GoogleSearchTool
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("agent.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Verify API key
if not os.getenv('GOOGLE_API_KEY'):
    raise ValueError("GOOGLE_API_KEY not found")

# Custom tool
def get_current_time() -> str:
    """Returns the current date and time"""
    return f"Current time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

# Callbacks for logging
def log_before_agent(ctx):
    logger.info(f"Agent: {ctx.agent.name}")

# Search agent
search_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='SearchAgent',
    instruction="""You are a specialist in Google Search grounding.
    Use web search to find current information and cite sources.""",
    tools=[GoogleSearchTool()],
    before_agent_callback=log_before_agent,
)

# Main agent
main_agent = LlmAgent(
    model='gemini-2.5-flash',
    name='MainAgent',
    instruction="""You are a helpful AI assistant.
    When asked about CURRENT events or RECENT information, 
    use search to provide accurate, source-attributed responses.""",
    tools=[
        AgentTool(agent=search_agent),
        FunctionTool(get_current_time),
    ],
    before_agent_callback=log_before_agent,
)

# Setup runner
session_service = InMemorySessionService()
runner = Runner(
    agent=main_agent,
    session_service=session_service,
    app_name="search_agent_app"
)

async def run_query(query: str):
    """Run a single query"""
    print(f"\n{'='*60}")
    print(f"🔍 QUERY: {query}")
    print(f"{'='*60}\n")

    session = session_service.create_session_sync()
    response_text = ""

    async for event in runner.run_async(
        user_message=query,
        session_id=session.session_id
    ):
        if hasattr(event, 'content') and event.content:
            for part in event.content.parts:
                if hasattr(part, 'text') and part.text:
                    response_text += part.text

    print(f"RESPONSE:\n{response_text}\n")
    print(f"{'-'*60}\n")

async def main():
    """Run test suite"""
    print("\n" + "="*60)
    print("ADK Search-Grounded Agent - Test Suite")
    print("="*60 + "\n")

    await run_query("What are the latest AI developments this week?")
    await run_query("What's today's top tech news?")
    await run_query("What's the current time?")

    print("\nTests completed!\n")

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python agent.py

What's Next?

🎉 Congratulations! You've built an AI agent with real-time search grounding!

Coming in Part 2: Google Maps Grounding

We'll add location intelligence to your agent:

Set up Vertex AI for Maps grounding
Handle location queries: "Find restaurants near Times Square"
Get directions and distances
Combine search and maps in one agent

The Complete 5-Part Series

✅ Part 1: Google Search Grounding (completed!)
Part 2: Google Maps Grounding with Vertex AI
Part 3: Full-Stack Frontend with CopilotKit & AG-UI
Part 4: Persistent Sessions with PostgreSQL & Docker
Part 5: Production Deployment on Cloud Run

By Part 5, you'll have a production-ready, scalable AI agent deployed to the cloud with persistent sessions and a beautiful UI!

Key Takeaways

✅ Search grounding solves the knowledge cutoff problem

✅ Agent-as-Tool pattern is mandatory for grounding tools

✅ Never mix GoogleSearchTool() directly with FunctionTools

✅ Dedicated agents for each grounding tool prevent conflicts

✅ Clear instructions tell the agent when to search

✅ Grounding metadata provides transparency and trust

Resources

📦 Full Code: https://github.com/Greyisheep/ag-ui-adk-grounding-app
📚 ADK Documentation: https://google.github.io/adk-docs/
🔍 Search Grounding Guide: https://cloud.google.com/vertex-ai/generative-ai/docs/grounding/grounding-with-google-search
🔑 Get API Key: https://makersuite.google.com/app/apikey

Questions?

Drop a comment below! I spent 3 hours debugging the "Function search is not found" error, so I'm happy to help you skip that pain 😅

⭐ Star the repo if this helped you, and follow me for Part 2 where we add Google Maps grounding!

About the Author: I'm Claret, an AI Engineer building production agentic systems. I lead a study group of 30+ engineers exploring AI agents and system design.

Connect with me on LinkedIn | GitHub

Mastering Google ADK DatabaseSessionService and Events: Complete Guide to Event Injection and Extraction

Claret Ibeawuchi — Fri, 26 Sep 2025 06:28:37 +0000

Introduction

Building sophisticated conversational AI systems requires robust event management. Google's Agent Development Kit (ADK) provides powerful tools for managing conversation state, but mastering event injection and extraction patterns is crucial for creating production-ready applications.

In this comprehensive guide, we'll explore real-world patterns from a production AI system, covering everything from basic event injection to advanced extraction techniques.

Understanding ADK DatabaseSessionService
Event Injection Patterns
Event Extraction Patterns
Real-World Implementation Examples
Best Practices and Common Pitfalls
Performance Considerations
Conclusion

Understanding Events and Sessions in ADK

Before going into injection and extraction patterns, it's crucial to understand what events are and how they relate to sessions in the Google ADK ecosystem.

What Are Events?

Events are the fundamental units of information flow within the Agent Development Kit (ADK). According to the official ADK documentation, events represent every significant occurrence during an agent's interaction lifecycle, from initial user input to the final response and all the steps in between.

An Event in ADK is an immutable record representing a specific point in the agent's execution. It captures:

User messages
Agent replies
Tool requests (function calls)
Tool results
State changes
Control signals
Errors

Event Structure

Based on the ADK Event class, events have this conceptual structure:

Why Events Matter

Events are central to ADK's operation for several key reasons:

Communication: They serve as the standard message format between the user interface, the Runner, agents, the LLM, and tools. Everything flows as an Event.
Signaling State & Artifact Changes: Events carry instructions for state modifications and track artifact updates. The SessionService uses these signals to ensure persistence via event.actions.state_delta and event.actions.artifact_delta.
Control Flow: Specific fields like event.actions.transfer_to_agent or event.actions.escalate act as signals that direct the framework, determining which agent runs next or if a loop should terminate.
History & Observability: The sequence of events recorded in session.events provides a complete, chronological history of an interaction, invaluable for debugging, auditing, and understanding agent behavior step-by-step.

Understanding Sessions

Sessions are containers that hold the complete conversation state. The DatabaseSessionService is the core component that manages conversation state in ADK applications. It handles:

Session Management: Creating, retrieving, and managing conversation sessions
Event Storage: Persisting conversation events with full history
State Management: Maintaining session state across interactions
Event Processing: Enabling real-time event injection and extraction

Event-Session Relationship

The relationship between events and sessions is fundamental:

Events are stored in sessions: Each session maintains a chronological list of all events that have occurred
Sessions provide context: Events can reference and modify session state through EventActions
Sessions enable persistence: The DatabaseSessionService ensures events are persisted and can be retrieved later
Events drive session evolution: Each event can modify session state, creating a living record of the conversation

Key Components

from google.adk.sessions import DatabaseSessionService
from google.adk.events import Event, EventActions
from google.genai import types

The EventActions class is particularly important as it contains the side effects and control signals that events can carry.

Event Flow and Processing

Understanding how events flow through the ADK system is crucial for effective event management. According to the ADK documentation, events follow this processing flow:

Generation: Events are created at different points:
- User Input: The Runner wraps initial user messages into an Event with author='user'
- Agent Logic: Agents explicitly yield Event(...) objects with author=self.name
- LLM Responses: The ADK model integration layer translates raw LLM output into Event objects
- Tool Results: After tool execution, the framework generates an Event containing the function_response
Processing Flow:
- Yield/Return: An event is generated and yielded by its source
- Runner Receives: The main Runner executing the agent receives the event
- SessionService Processing: The Runner sends the event to the configured SessionService
- Apply Deltas: The service merges event.actions.state_delta into session.state
- Persist to History: Appends the processed event to the session.events list
- External Yield: The Runner yields the processed event to the calling application

This flow ensures that state changes and history are consistently recorded alongside the communication content of each event.

Event Types and Identification

Based on the ADK documentation, you can identify event types by checking:

Who sent it? (event.author): 'user' for end-user input, agent name for agent output
What's the main payload? (event.content and event.content.parts):
- Text: Conversational messages
- Tool Call Request: LLM asking to execute tools
- Tool Result: Results from tool execution
Is it streaming? (event.partial): Indicates incomplete chunks of text

Event Injection Patterns

1. Direct Event Injection with `append_event()`

This is the most fundamental pattern for injecting events into a session. The append_event() method is part of the DatabaseSessionService and is perfect for system notifications and background processing.

Basic Implementation

async def inject_system_notification(
    session_id: str, 
    user_id: str, 
    notification_data: dict
) -> None:
    """Inject a system notification into an existing session"""

    # Initialize service
    svc = DatabaseSessionService(db_url=settings.sql_database_url)
    session = await svc.get_session(
        app_name="my_ai_app", 
        user_id=user_id, 
        session_id=session_id
    )

    # Create structured notification content
    system_content = f"[SYSTEM_NOTIFICATION]\n{json.dumps(notification_data, default=str)}"

    # Build the event
    sys_event = Event(
        invocation_id=f"sys_notif_{int(time.time())}",
        author="system",
        content=types.Content(
            role="system", 
            parts=[types.Part(text=system_content)]
        ),
        actions=EventActions()
    )

    # Inject the event
    await svc.append_event(session=session, event=sys_event)

Advanced: Background Processing Events

async def process_order_created(
    order_id: str, 
    session_id: str, 
    offer_id: str, 
    user_id: str
) -> None:
    """Process order creation with agent response"""

    svc = DatabaseSessionService(db_url=settings.sql_database_url)
    session = await svc.get_session(
        app_name="my_ai_app", 
        user_id=user_id, 
        session_id=session_id
    )

    # 1. Inject system event with order data
    order_data = await api_service.fetch_order_details(order_id, offer_id)
    system_content = "[ORDER.CREATED]\n" + json.dumps(order_data, default=str)

    sys_event = Event(
        invocation_id=f"bkgd_sys_{order_id}",
        author="system",
        content=types.Content(
            role="system", 
            parts=[types.Part(text=system_content)]
        ),
        actions=EventActions()
    )
    await svc.append_event(session=session, event=sys_event)

    # 2. Generate agent response
    prompt = "Summarize next steps based on the above order data."
    result = await non_streaming_agent.invoke(
        content=types.Content(
            role="user", 
            parts=[types.Part(text=prompt)]
        )
    )

    assistant_text = next(
        (p.text for p in result.parts if getattr(p, "text", None)), 
        None
    )

    if assistant_text:
        # 3. Inject agent response
        assistant_event = Event(
            invocation_id=f"bkgd_asst_{order_id}",
            author="assistant",
            content=types.Content(
                role="assistant", 
                parts=[types.Part(text=assistant_text)]
            ),
            actions=EventActions()
        )
        await svc.append_event(session=session, event=assistant_event)

When to use:

System notifications
Background data processing
Automated agent responses
Data synchronization events

2. State Management with State Deltas

Update session state without visible content using the EventActions.state_delta field. This pattern leverages the EventActions class and is essential for maintaining session metadata and user context.

Authentication State Management

async def inject_admin_authentication(
    session: Session,
    svc: DatabaseSessionService,
    auth_token: str,
    admin_user_id: str,
    customer_user_id: str = None
) -> None:
    """Inject admin authentication state into session"""

    # Build state delta
    state_delta = {
        "session:auth_token": auth_token,
        "session:admin_user_id": admin_user_id,
        "session:admin_mode": True
    }

    if customer_user_id:
        state_delta["session:customer_user_id"] = customer_user_id

    # Create state update event
    admin_auth_evt = Event(
        invocation_id="inject_admin_auth",
        author="system",
        actions=EventActions(state_delta=state_delta)
    )

    await svc.append_event(session=session, event=admin_auth_evt)

User Preference Management

async def update_user_preferences(
    session: Session,
    svc: DatabaseSessionService,
    timezone: str = None,
    offset_minutes: int = None
) -> None:
    """Update user timezone and offset preferences"""

    state_delta = {}
    if timezone is not None:
        state_delta["user:time_zone"] = timezone
    if offset_minutes is not None:
        state_delta["user:offset_minutes"] = int(offset_minutes)

    if state_delta:
        actions = EventActions(state_delta=state_delta)
        tz_evt = Event(
            invocation_id=str(uuid.uuid4()), 
            author="system", 
            actions=actions, 
            timestamp=time.time()
        )
        await svc.append_event(session=session, event=tz_evt)

When to use:

Authentication state updates
User preference management
Session metadata storage
Temporary session data

3. Chat Service Integration

For user-facing notifications that appear in the chat UI. This pattern bridges ADK sessions with chat interfaces.

Status Notifications

async def process_status_notification(
    session_id: str,
    user_id: str,
    order_id: str,
    event_type: str,
    status_data: StatusNotificationData,
    admin_user: str
) -> None:
    """Process status notification and inject into chat session"""

    try:
        # Create formatted message
        message_content = _format_status_message(event_type, status_data, order_id)

        # Validate session
        is_valid, error_msg = _validate_session_for_notification(db, session_id, user_id)
        if not is_valid:
            logger.error(f"Session validation failed: {error_msg}")
            return

        # Inject via chat service
        chat_service = ChatService(db)
        status_dict = status_data.model_dump()

        # Ensure datetime serialization
        for key, value in status_dict.items():
            if isinstance(value, datetime):
                status_dict[key] = value.isoformat()

        await chat_service.inject_system_message(
            session_id=session_id,
            user_id=user_id,
            message_content=message_content,
            message_type="status_notification",
            metadata={
                "order_id": order_id,
                "event_type": event_type,
                "admin_user": admin_user,
                "timestamp": datetime.utcnow().isoformat(),
                "status_data": status_dict
            }
        )

    except Exception as e:
        logger.error(f"Failed to inject status notification: {e}")
        # Don't raise - webhook should still succeed

Payment Notifications

async def inject_payment_notification(
    session_id: str,
    user_email: str,
    payment_type: str,
    amount: float,
    reference: str
) -> None:
    """Inject payment notification into chat session"""

    message = f"""
    💳 Payment {payment_type.title()} Notification

    Amount: ${amount:.2f}
    Reference: {reference}

    Your payment has been processed successfully!
    """

    try:
        chat_service = ChatService()
        await chat_service.inject_system_message(
            session_id=session_id,
            user_id=user_email,
            message_content=message,
            message_type="payment_notification",
            metadata={
                "payment_type": payment_type,
                "amount": amount,
                "reference": reference,
                "timestamp": datetime.utcnow().isoformat()
            }
        )
    except Exception as e:
        logger.error(f"Failed to inject payment notification: {e}")

When to use:

User-facing notifications
Status updates
Payment alerts
System status messages

Event Extraction Patterns

1. Direct Session Access

The simplest way to access all events in a session.

async def get_session_history(
    user_id: str, 
    session_id: str
) -> List[Event]:
    """Retrieve complete session history"""

    svc = DatabaseSessionService(db_url=settings.sql_database_url)
    session = await svc.get_session(
        app_name="my_ai_app", 
        user_id=user_id, 
        session_id=session_id
    )

    return session.events  # Complete event history

2. Intelligent Text Extraction

Process agent responses safely with proper error handling. This pattern is based on the ADK documentation for identifying and extracting text content from events.

def _extract_text_from_event(evt) -> str | None:
    """
    Safely extract the first text part from an ADK event.

    Handles cases where evt.content might be None and provides
    robust error handling for production systems.
    """
    content = getattr(evt, "content", None)
    if content is None:
        return None

    parts = getattr(content, "parts", None)
    if not parts:
        return None

    first_part = parts[0]

    # Skip function calls - they are handled internally by the runner
    if getattr(first_part, "function_call", None) is not None:
        return None

    return getattr(first_part, "text", None) or None

# Usage in agent response processing
async def process_agent_response(invocation_events: List[Event]) -> str:
    """Process agent response events and extract text content"""

    response_parts = []
    content = None

    try:
        for i, evt in enumerate(invocation_events):
            maybe_text = _extract_text_from_event(evt)
            if maybe_text:
                response_parts.append(maybe_text)
                content = maybe_text  # Keep the last non-empty response
                logger.debug(f"Extracted text from event {i}: {maybe_text[:100]}...")

        logger.info(f"Response extraction completed: {len(response_parts)} text parts found")

    except Exception as extraction_error:
        logger.error(f"Response extraction failed: {extraction_error}", exc_info=True)
        if response_parts:
            content = response_parts[-1]  # Use last successful extraction

    if content is None:
        raise ValueError("No response generated by agent")

    return content

3. Specialized Event Extraction

Extract specific event types like payment events, errors, or custom data structures.

Payment Event Extraction

def _extract_payment_event_from_event(evt) -> dict | None:
    """Extract payment events from agent tool execution events"""
    try:
        content = getattr(evt, "content", None)
        if content is None:
            return None

        parts = getattr(content, "parts", None)
        if not parts:
            return None

        # Look for function result parts that might contain payment events
        for part in parts:
            function_result = getattr(part, "function_result", None)
            if function_result is None:
                continue

            # Get the result content
            result_content = getattr(function_result, "content", None)
            if result_content is None:
                continue

            # Parse the result as JSON to check for payment events
            try:
                if isinstance(result_content, str):
                    result_data = json.loads(result_content)
                elif isinstance(result_content, dict):
                    result_data = result_content
                else:
                    continue

                # Check if this result contains a payment event
                if isinstance(result_data, dict) and result_data.get('payment_event'):
                    payment_event = result_data['payment_event']
                    if payment_event.get('type') == 'open_payment_modal':
                        return payment_event

            except (json.JSONDecodeError, AttributeError):
                continue

        return None

    except Exception as e:
        logger.debug(f"Error extracting payment event: {e}")
        return None

# Usage in payment processing
async def process_payment_events(invocation_events: List[Event]) -> dict | None:
    """Process agent events and extract payment information"""

    payment_event = None

    try:
        for evt in invocation_events:
            payment_evt = _extract_payment_event_from_event(evt)
            if payment_evt:
                payment_event = payment_evt
                logger.info(f"Payment event detected: {payment_evt.get('type', 'unknown')}")
                break

    except Exception as e:
        logger.error(f"Payment event processing failed: {e}")

    return payment_event

Detecting Final Responses

According to the ADK documentation, you can use the built-in event.is_final_response() method to identify events suitable for display as the agent's complete output:

async def process_agent_events(invocation_events: List[Event]) -> str:
    """Process agent events and extract final response"""

    full_response_text = ""
    final_response = None

    for event in invocation_events:
        # Accumulate streaming text if needed
        if event.partial and event.content and event.content.parts and event.content.parts[0].text:
            full_response_text += event.content.parts[0].text

        # Check if it's a final, displayable event
        if event.is_final_response():
            if event.content and event.content.parts and event.content.parts[0].text:
                # Use accumulated text for final response
                final_response = full_response_text + (event.content.parts[0].text if not event.partial else "")
                break

    return final_response or "No final response generated"

When is_final_response() returns True:

The event contains a tool result with skip_summarization=True
The event contains a tool call for a long-running tool
All of the following are met:
- No function calls
- No function responses
- Not a partial stream chunk
- Doesn't end with code execution results

Error Event Extraction

def _extract_error_from_event(evt) -> dict | None:
    """Extract error information from agent events"""
    try:
        content = getattr(evt, "content", None)
        if content is None:
            return None

        parts = getattr(content, "parts", None)
        if not parts:
            return None

        for part in parts:
            # Check for error indicators
            if hasattr(part, 'text') and part.text:
                text = part.text.lower()
                if any(keyword in text for keyword in ['error', 'failed', 'exception']):
                    return {
                        'type': 'error',
                        'message': part.text,
                        'timestamp': getattr(evt, 'timestamp', None)
                    }

    except Exception as e:
        logger.debug(f"Error extracting error event: {e}")

    return None

Real-World Implementation Examples

Session Migration Pattern

async def migrate_anonymous_session(
    svc: DatabaseSessionService,
    anonymous_session_id: str,
    authenticated_user_id: str
) -> Session:
    """Migrate anonymous session to authenticated user"""

    # Get anonymous session
    anonymous_session = await svc.get_session(
        app_name="my_ai_app", 
        user_id="anonymous", 
        session_id=anonymous_session_id
    )

    if not anonymous_session:
        raise ValueError("Anonymous session not found")

    # Get all events from anonymous session
    anonymous_events = anonymous_session.events

    # Create new session for authenticated user
    session = await svc.create_session(
        app_name="my_ai_app",
        user_id=authenticated_user_id,
        session_id=anonymous_session_id  # Keep same session ID
    )

    # Migrate events (filtering out system events that will be recreated)
    migrated_events = 0
    if anonymous_events:
        for event in anonymous_events:
            # Skip system events that will be recreated
            if event.author == "system" and event.invocation_id.startswith("sys_"):
                continue

            # Re-inject the event into the new session
            await svc.append_event(session=session, event=event)
            migrated_events += 1

    logger.info(f"Migrated {migrated_events} events from anonymous to authenticated session")
    return session

Real-time Event Processing

async def process_live_events(live_events):
    """Process live agent events in real-time"""

    async for event in live_events:
        try:
            # Extract text content
            text_content = _extract_text_from_event(event)
            if text_content:
                yield {
                    "type": "text",
                    "content": text_content,
                    "timestamp": getattr(event, 'timestamp', time.time())
                }

            # Extract payment events
            payment_event = _extract_payment_event_from_event(event)
            if payment_event:
                yield {
                    "type": "payment",
                    "content": payment_event,
                    "timestamp": getattr(event, 'timestamp', time.time())
                }

        except Exception as e:
            logger.error(f"Error processing live event: {e}")
            yield {
                "type": "error",
                "content": str(e),
                "timestamp": time.time()
            }

Best Practices and Common Pitfalls

1. Session Validation

Always validate sessions before injection:

def _validate_session_for_notification(
    db: Session, 
    session_id: str, 
    user_id: str
) -> Tuple[bool, Optional[str]]:
    """Validate session exists and belongs to user"""

    try:
        # Check if session exists in chat system
        room = db.query(ChatRoom).filter(
            ChatRoom.id == UUID(session_id),
            ChatRoom.user_id == user_id
        ).first()

        if not room:
            return False, f"Session {session_id} not found for user {user_id}"

        return True, None

    except Exception as e:
        return False, f"Session validation error: {str(e)}"

2. Error Handling

Implement robust error handling to prevent system failures:

async def safe_event_injection(
    svc: DatabaseSessionService,
    session: Session,
    event: Event
) -> bool:
    """Safely inject event with error handling"""

    try:
        await svc.append_event(session=session, event=event)
        return True
    except Exception as e:
        logger.error(f"Failed to inject event: {e}")
        # Don't raise - allow system to continue
        return False

3. Unique Invocation IDs

Use unique invocation IDs to prevent duplicate events:

def generate_unique_invocation_id(prefix: str) -> str:
    """Generate unique invocation ID with timestamp and UUID"""
    timestamp = int(time.time())
    uuid_suffix = str(uuid.uuid4())[:8]
    return f"{prefix}_{timestamp}_{uuid_suffix}"

4. Event Ordering

Maintain proper event ordering:

async def inject_ordered_events(
    svc: DatabaseSessionService,
    session: Session,
    events: List[Event]
) -> None:
    """Inject events in proper order"""

    # Sort events by timestamp if available
    sorted_events = sorted(
        events, 
        key=lambda e: getattr(e, 'timestamp', 0)
    )

    for event in sorted_events:
        await svc.append_event(session=session, event=event)

Performance Considerations

1. Batch Operations

For multiple events, consider batch operations:

async def batch_event_injection(
    svc: DatabaseSessionService,
    session: Session,
    events: List[Event]
) -> None:
    """Inject multiple events efficiently"""

    # Process events in batches to avoid overwhelming the system
    batch_size = 10
    for i in range(0, len(events), batch_size):
        batch = events[i:i + batch_size]

        # Process batch concurrently
        tasks = [
            svc.append_event(session=session, event=event)
            for event in batch
        ]
        await asyncio.gather(*tasks, return_exceptions=True)

2. Event Filtering

Filter events during extraction to improve performance:

def filter_recent_events(events: List[Event], hours: int = 24) -> List[Event]:
    """Filter events to only recent ones"""

    cutoff_time = time.time() - (hours * 3600)

    return [
        event for event in events
        if getattr(event, 'timestamp', 0) > cutoff_time
    ]

3. Memory Management

For large session histories, implement pagination:

async def get_paginated_events(
    svc: DatabaseSessionService,
    user_id: str,
    session_id: str,
    page: int = 0,
    page_size: int = 100
) -> List[Event]:
    """Get paginated session events"""

    session = await svc.get_session(
        app_name="my_ai_app",
        user_id=user_id,
        session_id=session_id
    )

    all_events = session.events
    start_idx = page * page_size
    end_idx = start_idx + page_size

    return all_events[start_idx:end_idx]

Conclusion

Mastering ADK DatabaseSessionService event injection and extraction is crucial for building production-ready conversational AI systems. The patterns covered in this guide provide a solid foundation for:

System Integration: Seamlessly integrating with external systems
Real-time Processing: Handling live events and notifications
State Management: Maintaining conversation context
Error Handling: Building robust, fault-tolerant systems

Key Takeaways

Choose the right injection pattern for your use case
Implement robust error handling to prevent system failures
Use unique invocation IDs to avoid duplicate events
Consider performance implications for large-scale deployments
Validate sessions before injection operations

Next Steps

Experiment with the patterns in your own projects
Consider implementing monitoring for event injection/extraction
Explore advanced patterns like event streaming and real-time processing
Share your experiences and learnings with the community

The ADK DatabaseSessionService is a powerful tool when used correctly. These patterns will help you build sophisticated conversational AI systems that can handle real-world complexity while maintaining reliability and performance.

References

This guide is based on the official Google ADK documentation and source code. Here are the key references used:

Official Documentation

ADK Events Documentation - Comprehensive guide to understanding events in ADK
Google ADK Python Repository - Official ADK Python SDK

Source Code References

Event Class - Core Event implementation
EventActions Class - EventActions for state deltas and control signals
DatabaseSessionService - Session management and event persistence
Session Class - Session data structure
BaseSessionService - Base session service interface

Key Concepts Referenced

Event Structure: Based on the Event class extending LlmResponse with ADK-specific fields
Event Flow: Processing flow from generation to persistence as documented in ADK docs
State Management: Using EventActions.state_delta for session state updates
Session Persistence: How DatabaseSessionService.append_event() handles event storage
Event Identification: Methods for identifying event types as described in ADK documentation

Additional Resources

ADK GitHub Repository - Full source code and examples
ADK Documentation Hub - Complete ADK documentation
ADK Python SDK - PyPI package for installation

Have you implemented similar patterns in your projects? Share your experiences and any additional insights in the comments below!

Clean, Performant, and Testable: Mastering Data Access in Go with Repositories & sqlc (2)

Claret Ibeawuchi — Sun, 25 May 2025 22:34:39 +0000

Part 2: Implementing Effective Repositories in Go & Confronting the Critiques

(Recap: In Part 1, we explored the challenges of data access in Go, defined the Repository Pattern, and compared it to raw SQL and ORMs, setting the stage for a more structured approach. If you missed it, you can catch up here.)

Table of Contents (Part 2)

The 3 AM Database Panic
Code Deep Dive: Building a Manual Go Repository
- Defining the Domain Model
- Crafting the Repository Interface
- The pgx Implementation: Step-by-Step
- Observations on the Manual Approach
Why This Changes Everything: The Concrete Advantages
- Advantage 1: Enhanced Separation of Concerns (SoC)
- Advantage 2: Dramatically Improved Testability
- Advantage 3: Increased Flexibility & Maintainability
- Advantage 4: Centralized Data Access Logic
The Honest Truth: Acknowledging the Downsides
- Disadvantage 1: Boilerplate Code
- Disadvantage 2: Potential for Over-Abstraction
Confronting the Critics: A Go Developer's Response
- Critique 1: Hiding Native ORM Features
- Critique 2: "Deep Abstractions Kill Flexibility"
- Critique 3: SOLID Violations
- Critique 4: Unit Testing Isn't That Much Easier
- Critique 5: Leaky Abstraction
- Critique 6: CQRS Doesn't Fit Well
- Critique 7: The "Swappability" Stories
- Critique 8: Performance Anti-Patterns
- The Verdict: Context Matters
The Path Forward: Setting Up for Success
What's Next: The sqlc Revolution

The 3 AM Database Panic

It's 3 AM. Your phone buzzes with that dreaded sound. Production is down.

"The payment service is throwing database errors," your teammate types frantically in Slack. "I can't figure out where the bug is, there are SQL queries scattered across twelve different files."

You've been here before. What started as a simple microservice has become a web of embedded SQL, making debugging feel like archaeological excavation. Each fix breaks something else. Each test requires spinning up a full database. Each new feature means copying and pasting similar queries across multiple handlers.

This is the moment when clean architecture stops being academic and starts being survival.

In Part 1, we explored the theoretical foundations of the Repository Pattern through our story of three teams tackling the same problem with different approaches. Now it's time to get our hands dirty. We're going to build a repository from scratch, understand exactly why it solves the 3 AM panic scenario, and address the skeptics who argue it's unnecessary complexity.

By the end of this article, you'll never again wonder if repositories are worth the effort, you'll wonder how you survived without them.

Code Deep Dive: Building a Manual Go Repository

To truly appreciate what tools like sqlc (which we'll cover in Part 3) bring to the table, it's essential to first understand how to build a repository manually. This exercise highlights both the structural benefits of the pattern and the boilerplate that sqlc aims to eliminate. We'll use pgx/v5 with a PostgreSQL database.

Let's build this step by step, just like you would in a real project.

Defining the Domain Model

First, let's define our application's representation of a user. This struct lives in your domain layer and represents your business entity, not your database table.

// models/user.go
package models

import "time"

// User represents a user in our application domain.
type User struct {
    ID        int64     `json:"id"`
    Name      string    `json:"name"`
    Email     string    `json:"email" binding:"required,email"` // Example with validation tags
    CreatedAt time.Time `json:"created_at"`
    UpdatedAt time.Time `json:"updated_at"`
}

💡 Pro Tip: Keep your domain models focused on business logic, not database concerns. The json tags are for API serialization, and binding tags can be used with validation libraries like gin's validator.

Crafting the Repository Interface

Next, we define the contract for our user data operations. This interface is what your service layer will depend on—and what makes testing a breeze.

// domain/repositories/user_repository.go
package repositories

import (
    "context"
    "yourapp/models" // Adjust path to your models package
)

// UserRepository defines the interface for user data operations.
// This is the contract that our service layer depends on.
type UserRepository interface {
    Create(ctx context.Context, user *models.User) error
    GetByEmail(ctx context.Context, email string) (*models.User, error)
    GetByID(ctx context.Context, id int64) (*models.User, error)
    // Future methods could include:
    // Update(ctx context.Context, user *models.User) error
    // Delete(ctx context.Context, id int64) error
    // ListActiveUsers(ctx context.Context, limit int) ([]models.User, error)
}

Key Design Decisions:

context.Context: Standard Go practice for cancellation, deadlines, and request-scoped values
*models.User for Create: Allows the repository to set database-generated fields like ID, CreatedAt, and UpdatedAt back onto the user object
(*models.User, error) for Getters: If a user isn't found, we'll return nil for the user and a domain-specific error

The `pgx` Implementation: Step-by-Step

Now comes the real work—implementing UserRepository using pgx to interact with PostgreSQL. This code would typically reside in your infrastructure/persistence layer.

// infrastructure/persistence/postgres/pgx_user_repository.go
package postgres

import (
    "context"
    "errors"
    "fmt"
    "log" // In a real app, use a structured logger like slog or zerolog

    "yourapp/models"              // Adjust to your models path
    "yourapp/domain/repositories" // Adjust to your interface path

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgconn"
    "github.com/jackc/pgx/v5/pgxpool"
)

// Custom domain-specific errors (can be defined in a shared errors package)
var (
    ErrUserNotFound    = errors.New("user not found")
    ErrUserEmailExists = errors.New("user with this email already exists")
)

// pgxUserRepository implements the UserRepository interface using pgx.
type pgxUserRepository struct {
    dbpool *pgxpool.Pool
}

// NewPgxUserRepository creates a new instance of pgxUserRepository.
func NewPgxUserRepository(dbpool *pgxpool.Pool) repositories.UserRepository {
    return &pgxUserRepository{dbpool: dbpool}
}

// Create inserts a new user into the database.
func (r *pgxUserRepository) Create(ctx context.Context, user *models.User) error {
    query := `
        INSERT INTO users (name, email)
        VALUES ($1, $2)
        RETURNING id, created_at, updated_at`

    // .Scan() will write the returned id, created_at, updated_at back into the user pointer
    err := r.dbpool.QueryRow(ctx, query, user.Name, user.Email).Scan(
        &user.ID,
        &user.CreatedAt,
        &user.UpdatedAt,
    )

    if err != nil {
        var pgErr *pgconn.PgError
        if errors.As(err, &pgErr) {
            // PostgreSQL error code for unique_violation (e.g., duplicate email)
            // See: https://www.postgresql.org/docs/current/errcodes-appendix.html
            if pgErr.Code == "23505" {
                return fmt.Errorf("%w: %s", ErrUserEmailExists, user.Email)
            }
        }
        log.Printf("PgxUserRepository.Create: failed to insert user %s: %v", user.Email, err)
        return fmt.Errorf("failed to create user: %w", err)
    }
    return nil
}

// GetByEmail retrieves a user by their email address.
func (r *pgxUserRepository) GetByEmail(ctx context.Context, email string) (*models.User, error) {
    query := `
        SELECT id, name, email, created_at, updated_at
        FROM users
        WHERE email = $1`

    var user models.User
    err := r.dbpool.QueryRow(ctx, query, email).Scan(
        &user.ID,
        &user.Name,
        &user.Email,
        &user.CreatedAt,
        &user.UpdatedAt,
    )

    if err != nil {
        if errors.Is(err, pgx.ErrNoRows) {
            return nil, ErrUserNotFound // Return our domain-specific error
        }
        log.Printf("PgxUserRepository.GetByEmail: failed to get user %s: %v", email, err)
        return nil, fmt.Errorf("failed to get user by email: %w", err)
    }
    return &user, nil
}

// GetByID retrieves a user by their unique ID.
func (r *pgxUserRepository) GetByID(ctx context.Context, id int64) (*models.User, error) {
    query := `
        SELECT id, name, email, created_at, updated_at
        FROM users
        WHERE id = $1`

    var user models.User
    err := r.dbpool.QueryRow(ctx, query, id).Scan(
        &user.ID,
        &user.Name,
        &user.Email,
        &user.CreatedAt,
        &user.UpdatedAt,
    )

    if err != nil {
        if errors.Is(err, pgx.ErrNoRows) {
            return nil, ErrUserNotFound // Domain-specific error
        }
        log.Printf("PgxUserRepository.GetByID: failed to get user with id %d: %v", id, err)
        return nil, fmt.Errorf("failed to get user by id: %w", err)
    }
    return &user, nil
}

// Compile-time check to ensure pgxUserRepository satisfies UserRepository.
var _ repositories.UserRepository = (*pgxUserRepository)(nil)

Observations on the Manual Approach

Look at what we just built. Even for these simple CRUD operations, several patterns emerge:

SQL Strings in Go Code: Queries are embedded as strings. This is prone to typos and syntax errors only caught at runtime
Manual Scan() Calls: The order and number of arguments in Scan() must exactly match the columns in the SELECT statement. Any discrepancy leads to runtime errors or subtle data corruption
Error Handling Dance: We must explicitly check for pgx.ErrNoRows and map it to our domain-specific ErrUserNotFound. Handling other database errors (like unique constraint violations) also requires specific PostgreSQL knowledge
Repetitive Boilerplate: Even for these simple operations, there's noticeable repetitive code for query execution, scanning, and error mapping

This manual implementation gives us the architectural benefits of the Repository Pattern, but the implementation details are verbose and error-prone. This is exactly why sqlc will feel like magic in Part 3.

Why This Changes Everything: The Concrete Advantages

"But wait," you might say, "that's a lot of code for simple database operations. Why not just put the SQL directly in my handlers?"

Let me show you why this seemingly extra work transforms how you build and maintain Go applications.

Advantage 1: Enhanced Separation of Concerns (SoC)

Remember our 3 AM debugging scenario? Here's the architecture that prevents it:

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   HTTP Handler  │    │  Service Layer  │    │   Repository    │
│                 │───▶│                 │───▶│   Interface     │
│ • Parse Request │    │ • Business Logic│    │ • GetByID()     │
│ • Validate      │    │ • Orchestration │    │ • Create()      │
│ • Serialize     │    │ • Error Handling│    │ • GetByEmail()  │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │ PostgreSQL Impl │
                                               │                 │
                                               │ • SQL Queries   │
                                               │ • Error Mapping │
                                               │ • Type Scanning │
                                               └─────────────────┘

Aspect of SoC	Benefit	3 AM Impact
Business Logic Isolation	Service layers focus purely on business rules, unaware of persistence details	Bug in user creation? Check the service. SQL error? Check the repository.
Data Access Encapsulation	All database interaction logic is contained within the repository	One place to look for all user-related SQL queries
Independent Evolution	Changes to database schema primarily affect the repository, not the entire service layer	Schema change affects 1 file, not 12 scattered handlers
Clearer Team Responsibilities	Different developers can focus on business logic vs. data access optimization	Junior dev works on business logic, senior dev optimizes queries

Advantage 2: Dramatically Improved Testability

This is where repositories truly shine. Because services depend on repository interfaces, unit testing becomes clean and focused.

The Magic: Interface-Based Testing

// services/user_service_test.go
package services_test

import (
    "context"
    "testing"
    "time"
    "yourapp/models"
    "yourapp/domain/repositories"
    "yourapp/services"
    "errors"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/mock"
)

// MockUserRepository - our test double
type MockUserRepository struct {
    mock.Mock
}

func (m *MockUserRepository) Create(ctx context.Context, user *models.User) error {
    args := m.Called(ctx, user)
    // Simulate DB setting fields
    if args.Error(0) == nil {
        user.ID = 123
        user.CreatedAt = time.Now()
        user.UpdatedAt = time.Now()
    }
    return args.Error(0)
}

func (m *MockUserRepository) GetByEmail(ctx context.Context, email string) (*models.User, error) {
    args := m.Called(ctx, email)
    if user, ok := args.Get(0).(*models.User); ok {
        return user, args.Error(1)
    }
    return nil, args.Error(1)
}

func (m *MockUserRepository) GetByID(ctx context.Context, id int64) (*models.User, error) {
    args := m.Called(ctx, id)
    if user, ok := args.Get(0).(*models.User); ok {
        return user, args.Error(1)
    }
    return nil, args.Error(1)
}

// Test the happy path
func TestUserService_Register_Success(t *testing.T) {
    mockRepo := new(MockUserRepository)
    userService := services.NewUserService(mockRepo) // Inject our mock

    testUser := &models.User{Email: "test@example.com", Name: "Test User"}

    // Configure the mock: when Create is called, return success
    mockRepo.On("Create", mock.AnythingOfType("*context.emptyCtx"), testUser).Return(nil)

    // Execute the service method
    createdUser, err := userService.RegisterNewUser(context.Background(), "Test User", "test@example.com")

    // Verify results
    assert.NoError(t, err)
    assert.Equal(t, int64(123), createdUser.ID) // ID set by mock
    mockRepo.AssertExpectations(t) // Verify Create was called
}

// Test the error path
func TestUserService_Register_EmailExists(t *testing.T) {
    mockRepo := new(MockUserRepository)
    userService := services.NewUserService(mockRepo)

    // Configure mock to return our domain error
    mockRepo.On("Create", mock.AnythingOfType("*context.emptyCtx"), mock.AnythingOfType("*models.User")).
        Return(repositories.ErrUserEmailExists)

    // Execute the service method
    _, err := userService.RegisterNewUser(context.Background(), "Test User", "existing@example.com")

    // Verify error handling
    assert.Error(t, err)
    assert.True(t, errors.Is(err, repositories.ErrUserEmailExists))
    mockRepo.AssertExpectations(t)
}

The Result: Tests that run in milliseconds, require no database setup, and test your business logic in isolation. When you're debugging at 3 AM, these fast, reliable tests become your best friend.

Advantage 3: Increased Flexibility & Maintainability

The interface-based approach opens up powerful possibilities:

Swap Implementations with Zero Service Changes:

// Production: PostgreSQL
userRepo := postgres.NewPgxUserRepository(dbpool)

// Testing: In-memory
userRepo := memory.NewInMemoryUserRepository()

// Development: SQLite for fast local setup
userRepo := sqlite.NewSQLiteUserRepository(sqliteDB)

// The service doesn't care which one you use
userService := services.NewUserService(userRepo)

Add Caching with the Decorator Pattern:

type cachingUserRepository struct {
    next  repositories.UserRepository
    cache *redis.Client
}

func (r *cachingUserRepository) GetByID(ctx context.Context, id int64) (*models.User, error) {
    // Check cache first
    if cached := r.getCachedUser(id); cached != nil {
        return cached, nil
    }

    // Fall back to database
    user, err := r.next.GetByID(ctx, id)
    if err == nil {
        r.cacheUser(user)
    }
    return user, err
}

Advantage 4: Centralized Data Access Logic

All SQL queries and data mapping logic for User entities live in one place. When you need to:

Update a query for performance
Handle a schema change
Add logging or metrics
Debug a data issue

You know exactly where to look. No more grep-ing through dozens of files hunting for that one SQL query.

The Honest Truth: Acknowledging the Downsides

I won't sugarcoat this, the manual repository implementation we just built isn't without its drawbacks. Let's be honest about the costs:

Disadvantage 1: Boilerplate Code

Look at our pgxUserRepository again. Even for simple CRUD operations, there's significant repetitive code:

Writing SQL query strings (error-prone)
Calling dbpool.QueryRow() or dbpool.Query()
Manually scanning results with row.Scan() (order-dependent, fragile)
Mapping between database rows and Go structs
Handling pgx.ErrNoRows consistently

This boilerplate is verbose and a common source of subtle bugs. Miss one field in a Scan() call? Runtime panic. Change the order of columns? Silent data corruption.

Disadvantage 2: Potential for Over-Abstraction

If you try to create overly generic repositories (like Save(entity interface{}) error), you lose type safety and create interfaces that are hard to use correctly.

The Go Way: Go encourages specificity over generics. Tailored interfaces like UserRepository with explicit methods like GetActiveUsersByRegion() are preferred. This actually aligns perfectly with how sqlc will generate code in Part 3.

These disadvantages, especially the boilerplate, are precisely what sqlc will help us solve. But first, let's address the critics.

Confronting the Critics: A Go Developer's Response

The Repository Pattern has vocal critics. In particular:

Mesut Ataşoy’s Medium article “The Dark Side of Repository Pattern: A Developer’s Honest Journey” 🔗 https://medium.com/@mesutatasoy/the-dark-side-of-repository-pattern-a-developers-honest-journey-eb51eba7e8d8
Anthony Alaribe’s X post “A challenge I have with the repository pattern…” 🔗 https://x.com/tonialaribe/status/1755636332141908207

Let’s address their concerns head-on with a Go-specific perspective.

Critique 1: Hiding Native ORM Features

The Claim:

“It hides native ORM features like change tracking, eager/lazy loading, and raw-SQL execution, forcing you to reinvent them poorly.”
— Mesut Ataşoy, The Dark Side of Repository Pattern

Go Reality: In our pgx implementation, we're not hiding an ORM—we're using a database driver directly. If PostgreSQL offers a feature accessible via pgx (like CopyFrom for bulk inserts), our repository can expose it through its interface.

// Nothing stops us from exposing PostgreSQL-specific features
type UserRepository interface {
    Create(ctx context.Context, user *models.User) error
    BulkCreate(ctx context.Context, users []models.User) error // Uses pgx.CopyFrom
    GetByID(ctx context.Context, id int64) (*models.User, error)
}

With sqlc (Part 3), this becomes even less of an issue since you write raw SQL, nothing is hidden.

Critique 2: Deep Abstractions Kill Flexibility

The Claim:

“Repositories often add unnecessary layers of abstraction, forcing you to re-implement filtering, eager-loading, and transaction management in your own code.”
— Mesut Ataşoy

Go Perspective: Go's philosophy discourages overly generic repositories. Instead of GetAll(filters interface{}), we build specific methods:

type UserRepository interface {
    GetActiveUsersByRegion(ctx context.Context, region string) ([]models.User, error)
    GetUsersRegisteredAfter(ctx context.Context, date time.Time) ([]models.User, error)
    GetUserWithOrderHistory(ctx context.Context, id int64) (*models.UserWithOrders, error)
}

Each method is backed by a specific, optimized SQL query. Transaction management happens at the service layer, with repositories participating explicitly.

Critique 3: SOLID Violations

The Claim:

“Many implementations violate SRP by mixing data access, mapping and querying logic in one class.”
— Mesut Ataşoy

Go Analysis:

SRP: Our pgxUserRepository has one responsibility—user data access via pgx. The mapping (scanning database rows to structs) is part of this data access concern.
Open/Closed & Interface Segregation: Go's focused interfaces help here. Add new queries by extending the interface, not modifying existing methods.
Dependency Inversion: Services depend on UserRepository (abstraction), not pgxUserRepository (concrete implementation). This is proper DIP.

Critique 4: Unit Testing Isn't That Much Easier

The Claim:

“You still have to mock complex repository behavior—better off spinning up an in-memory DB.”
— Mesut Ataşoy

Go Reality: Our testing example shows Go's interface-based mocking is straightforward and powerful for service-layer unit tests. Integration tests with real databases are complementary—they test different concerns:

Unit tests (with mocks): Test business logic, error handling, edge cases
Integration tests (with real DB): Test SQL correctness, schema compatibility, performance

Both are valuable. Repositories enable focused unit tests that were impossible with embedded SQL.

Critique 5: Leaky Abstraction

The Claim:

“You still end up handling transactions, DB-specific errors, performance tweaks—all through that repository layer.”
— Mesut Ataşoy

Go Perspective: With explicit pgx usage, database-specific features are conscious choices, not hidden surprises. We explicitly handle PostgreSQL error codes because we want that level of control.

// We choose what to expose vs. what to abstract
func (r *pgxUserRepository) Create(ctx context.Context, user *models.User) error {
    // We handle PG-specific error codes explicitly
    var pgErr *pgconn.PgError
    if errors.As(err, &pgErr) && pgErr.Code == "23505" {
        return ErrUserEmailExists // Abstract to domain error
    }
    // Other errors bubble up with context
    return fmt.Errorf("failed to create user: %w", err)
}

Critique 6: CQRS Doesn't Fit Well

The Claim:

“The repository pattern forces you to use the same interface for both reads and writes, making CQRS harder to implement cleanly.”
— Mesut Ataşoy

Go Solution: Define separate interfaces when implementing CQRS:

type UserCommandRepository interface {
    Create(ctx context.Context, user *models.User) error
    Update(ctx context.Context, user *models.User) error
    Delete(ctx context.Context, id int64) error
}

type UserQueryRepository interface {
    GetByID(ctx context.Context, id int64) (*models.User, error)
    GetByEmail(ctx context.Context, email string) (*models.User, error)
    SearchUsers(ctx context.Context, criteria SearchCriteria) ([]models.User, error)
}

The pattern doesn't prevent this separation—it enables it.

Critique 7: The "Swappability" Stories

The Claim:

“The biggest argument for repositories, IMO, is that you’re not sprinkling Active Record/Data Mapper all over your codebase
But then devs write tutorials and put ‘swappable persistence’ as the main advantage. Yikes.”
— @simioluwatomi

This Criticism is Valid—but misses the real benefits. The primary wins aren't about swapping PostgreSQL for MongoDB (which would require fundamental architecture changes anyway). The real benefits are:

Potential Daily Development Wins:

Testing: SQLite for fast tests, PostgreSQL for integration tests
Development: In-memory repositories for rapid prototyping
Environment Adaptation: Different connection pooling, caching strategies, or even managed database services

The Core Win: As the critic notes, the biggest advantage is not sprinkling data access logic throughout your business services. Centralized data logic is the real benefit, not hypothetical database swapping.

Critique 8: Performance Anti-Patterns

The Claim:

“You tend to make too many DB calls or fetch more data than needed because engineers reuse overly generic getters instead of writing use-case-specific queries.”
— Anthony Alaribe

“At scale, moving unneeded data over the network costs you latency and bandwidth. Making many separate DB calls which could have been a single query is expensive in terms of latency.”

This is a Critical Point, and represents an implementation anti-pattern, not a pattern flaw:

❌ Anti-Pattern (Don't Do This):

type UserRepository interface {
    GetUser(id int64, includes ...string) (*User, error) // Too generic!
}

// Results in inefficient reuse
user := userRepo.GetUser(123, "orders", "preferences", "activity_log") // Over-fetching
name := user.Name // Only needed the name!

✅ Best Practice (Do This):

type UserRepository interface {
    GetUserProfile(ctx context.Context, id int64) (*UserProfile, error)
    GetUserNameForDisplay(ctx context.Context, id int64) (string, error)
    GetUserWithRecentOrders(ctx context.Context, id int64) (*UserWithOrders, error)
}

Each method maps to a specific, optimized SQL query that fetches only what's needed. sqlc naturally encourages this approach since you write specific SQL for each generated method.

The Verdict: Context Matters

Many "dark side" arguments target heavy, generic ORM-backed repositories or stem from poor implementation practices. Go's philosophy of explicit interfaces, combined with tools like sqlc, significantly mitigates these concerns.

The key is:

Specific, purpose-built repository methods
backed by
Explicit, optimized SQL—exactly what we'll achieve in Part 3.

That’s how you keep your Go API clean, performant, and testable.

The Path Forward: Setting Up for Success

We've built a working repository, seen its benefits in action, and addressed the critics. Our manual pgxUserRepository demonstrates the architectural value of the pattern while highlighting the boilerplate burden.

This positions us perfectly for Part 3, where sqlc will eliminate the grunt work while preserving all the architectural benefits we've gained.

What's Next: The `sqlc` Revolution

In Part 3: sqlc to the Rescue! Building Type-Safe & Efficient Data Access in Go, we'll witness a transformation:

What We'll Cover:

sqlc Introduction: Philosophy, installation, and basic workflow
Dramatic Refactoring: Converting our manual repository to use sqlc-generated code
Advanced Features: Complex queries, transactions, custom types, and edge cases
Production Patterns: Error handling, testing strategies, and performance optimization
The Complete Picture: Bringing together repositories, sqlc, and Go best practices

The Payoff: By the end of Part 3, you'll have a data access layer that's:

✅ Type-safe: Compile-time guarantees instead of runtime panics
✅ Performant: Hand-tuned SQL with zero ORM overhead
✅ Maintainable: Clear separation of concerns with minimal boilerplate
✅ Testable: Easy mocking and focused unit tests
✅ Scalable: Patterns that grow with your application

Get ready to see how sqlc transforms Go database development from tedious to delightful.

What's your experience with data access patterns in Go? Have you tried repositories, or do you prefer embedding SQL directly in services? Share your thoughts and challenges in the comments—your insights help the entire Go community learn and grow together.

Series Navigation:

← Part 1: The Pursuit for Clean Data Access
Part 2: You are here
Part 3: Coming soon...

Clean, Performant, and Testable: Mastering Data Access in Go with Repositories & sqlc

Claret Ibeawuchi — Fri, 23 May 2025 09:31:53 +0000

Part 1: The Pursuit for Clean Data Access in Go: Understanding the Repository Pattern

Introduction: The Data Access Dilemma
What Is the Repository Pattern?
The Wild West: Life Without Repositories
The Allure of ORMs: Convenience vs. Control
Story Time: Three Teams, One Goal
The Path Forward
What's Next in This Series

Introduction: The Data Access Dilemma

By the end of this article, you'll understand why the Repository Pattern is essential for building maintainable, testable, and performant Go applications.

Picture this: You're three months into a Go project. What started as a simple service with a few database calls has grown into a complex system. Your HTTP handlers are bloated with SQL queries, your tests require a real database to run, and every schema change sends you on a hunt through dozens of files to update hardcoded queries.

Sound familiar? You're not alone.

In the world of backend development, especially with Go's emphasis on simplicity and performance, how we interact with databases is a critical architectural decision. It influences everything from code maintainability and testability to application performance and scalability.

This three-part series will guide you through a pragmatic and powerful approach: combining the time-tested Repository Pattern with sqlc, a modern tool that generates type-safe Go code from your SQL. By the end, you'll have the knowledge to build robust, clean, and efficient data access layers that scale with your application.

In this first part, we'll explore the foundational concepts, common pitfalls, and set the stage for the practical implementations to come.

What Is the Repository Pattern?

The Repository Pattern is a design pattern that creates an abstraction layer between your application's business logic and data access logic. Think of it as a specialized librarian who knows exactly where to find any book (data) you need, regardless of how the library is organized behind the scenes.

The Core Purpose: Decoupling

The Repository Pattern's primary goal is decoupling. Your application services interact with a well-defined interface, remaining blissfully unaware of:

The underlying data store (PostgreSQL, MySQL, MongoDB, or even in-memory collections)
How data is accessed (raw SQL, ORM, or sqlc)
Database-specific quirks and connection management

Key Characteristics

Characteristic	Description	Benefit
Interface-Driven	Data operations are defined by Go interfaces (e.g., `GetUserByID`, `CreateOrder`)	Enables easy mocking and testing
Encapsulation	Hides low-level database interaction details	Reduces complexity in business logic
Collection Semantics	Presents data as if it were an in-memory collection	Intuitive API for developers

The Payment Terminal Analogy

Imagine a cashier at a busy supermarket. The cashier (your service layer) focuses on checking out customers. When processing payment, they interact with a payment terminal (your repository interface) by simply selecting "Card Payment" and following prompts.

The cashier doesn't need to know:

Whether it's Visa, Mastercard, or Amex
Which bank processes the transaction
What encryption protocols are used
How the network communication works

The payment terminal (your repository implementation) handles all these complexities and returns a simple result: "Payment Approved" or "Payment Declined."

This abstraction is powerful. The store can switch payment processors, update security protocols, or change communication methods without retraining cashiers or modifying the checkout process.

High-Level Architecture

The Wild West: Life Without Repositories

Before we appreciate the Repository Pattern's elegance, let's examine the chaos of unstructured data access—what I call the "Wild West" approach.

The Tempting Simplicity

It's common, especially in smaller projects, to see database logic embedded directly in service functions or HTTP handlers. Here's what this looks like:

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

type User struct {
    ID        int64
    Name      string
    Email     string
    CreatedAt time.Time
}

type UserService struct {
    dbpool *pgxpool.Pool
}

func (s *UserService) CreateUserAndNotify(ctx context.Context, name, email string) (User, error) {
    var userID int64
    var createdAt time.Time

    // Direct SQL embedded in business logic
    query := "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id, created_at"
    err := s.dbpool.QueryRow(ctx, query, name, email).Scan(&userID, &createdAt)
    if err != nil {
        log.Printf("Error creating user %s: %v", email, err)
        return User{}, fmt.Errorf("could not create user: %w", err)
    }

    newUser := User{ID: userID, Name: name, Email: email, CreatedAt: createdAt}

    // Business logic mixed with data access
    // - Send welcome email
    // - Log audit event
    // - Update cache

    log.Printf("User '%s' created with ID %d", newUser.Name, newUser.ID)
    return newUser, nil
}

The Growing Pains

Problem	Impact	Example
Testing Nightmare	Can't unit test business logic without database	Mock `pgxpool.Pool` and all its methods
Responsibility Overload	Services handle business logic AND data access	Single function does SQL + validation + notifications
Code Duplication	Same queries scattered across files	`SELECT * FROM users WHERE id = $1` repeated everywhere
Fragile Refactoring	Schema changes require codebase-wide updates	Rename a column? Hunt through 50 files
Tight Coupling	Business logic tied to specific database/driver	Want to add caching? Major rewrite needed

The Maintenance Burden

As your application grows, this approach quickly becomes inconvenient:

Testing becomes impossible without complex database setup
Code reviews become exercises in spotting SQL typos
Performance optimization requires hunting through business logic
Transaction management becomes spaghetti code across functions

While direct database access might seem simpler initially, it often leads to what architects call a "Big Ball of Mud"—a system that's difficult to maintain, test, and evolve.

The Allure of ORMs: Convenience vs. Control

When developers encounter the pains of raw SQL mixed with business logic, Object-Relational Mappers (ORMs) often appear as a silver bullet. Popular Go ORMs include GORM, Bun, and SQLBoiler.

What ORMs Promise

ORMs aim to provide a higher-level abstraction for database interactions, mapping tables to Go structs and offering APIs for CRUD operations without raw SQL.

The Seductive Benefits

Benefit	Description	Example
Rapid Development	Quick CRUD operations	`db.Create(&Product{Code: "D42", Price: 100})`
Database Agnosticism	Switch databases with configuration	PostgreSQL to MySQL with a setting change
Less SQL	Object-oriented query building	`db.Where("price > ?", 100).Find(&products)`

The Go Philosophy Clash

However, ORMs come with trade-offs that can feel at odds with Go's emphasis on simplicity and explicitness:

1. The "Hidden Magic" Problem

// What SQL does this generate?
db.Preload("Orders.Items").Where("age > ?", 18).Find(&users)

// Could be efficient... or could be a performance disaster

2. Complex Abstractions

Learning a full-featured ORM often requires as much effort as learning SQL itself, plus:

Struct tagging conventions
Query builder syntax
Transaction management
Association handling
Performance optimization strategies

3. Performance Overheads

The abstraction layer introduces overhead that can become noticeable in high-throughput applications.

4. Leaky Abstractions

Despite promises of abstracting away the database, you still need deep SQL knowledge for:

Performance optimization
Complex queries
Debugging
Advanced database features

When ORMs Shine vs. When They Struggle

Scenario	ORM Fit	Reason
Admin Panels	✅ Excellent	Simple CRUD operations
Prototypes	✅ Good	Rapid development
Complex Analytics	❌ Poor	Need optimized SQL
High Performance	❌ Challenging	Overhead and unpredictability
Database-Specific Features	❌ Limited	Abstractions get in the way

Story Time: Three Teams, One Goal

To illustrate these different approaches in practice, let's follow three development teams building an invoicing application. Each team takes a different path up "Data Mountain."

The Teams

Team Alpha (Lead: Claret) - Chooses GORM
Team Beta (Lead: Musa) - Implements Repository Pattern manually
Team Gamma (Lead: Sofia) - Uses raw SQL in services

Act 1: The Easy Start

Team Alpha (GORM)

Claret's team hit the ground running:

type Invoice struct {
    gorm.Model
    ClientID uint
    Amount   float64
    Status   string
    Client   Client
    Items    []LineItem
}

// Magic happens
db.AutoMigrate(&Invoice{}, &Client{}, &LineItem{})
db.Create(&Invoice{ClientID: 1, Amount: 100.50, Status: "pending"})

"GORM is amazing!" Claret exclaimed. Basic operations were up in days.

Team Beta (Repository Pattern)

Musa's team took a more methodical approach:

type InvoiceRepository interface {
    Create(ctx context.Context, invoice Invoice) (Invoice, error)
    GetByID(ctx context.Context, id int64) (Invoice, error)
    GetByClientID(ctx context.Context, clientID int64) ([]Invoice, error)
}

type postgresInvoiceRepo struct {
    db *pgxpool.Pool
}

func (r *postgresInvoiceRepo) Create(ctx context.Context, invoice Invoice) (Invoice, error) {
    query := `INSERT INTO invoices (client_id, amount, status) 
              VALUES ($1, $2, $3) RETURNING id, created_at`
    // ... implementation
}

Setup took longer, but Musa felt confident: "Clean interfaces, explicit SQL, full control."

Team Gamma (Raw SQL)

Sofia embedded database calls directly:

func (h *InvoiceHandler) CreateInvoice(w http.ResponseWriter, r *http.Request) {
    // Parse request...

    var invoiceID int64
    err := h.dbpool.QueryRow(ctx, 
        "INSERT INTO invoices (client_id, amount) VALUES ($1, $2) RETURNING id",
        clientID, amount).Scan(&invoiceID)
    if err != nil {
        // Handle error...
    }

    // More business logic...
}

"No extra layers, pure Go and SQL," Sofia reasoned.

Act 2: The Complex Feature Challenge

The teams needed to implement "Year-End Financial Reporting" - a feature requiring complex aggregations, joins, and calculations best performed in the database.

Team Alpha's Struggle

// Trying to express complex SQL with GORM
result := db.Model(&Invoice{}).
    Select("SUM(CASE WHEN status = 'paid' AND paid_date BETWEEN ? AND ? THEN amount ELSE 0 END) as total_paid").
    Joins("JOIN clients ON invoices.client_id = clients.id").
    Where("clients.region = ?", region).
    Group("clients.region")

Claret frowned: "The generated SQL is inefficient, and debugging this chain of methods is painful."

Team Beta's Control

func (r *postgresReportRepo) GetYearEndReport(ctx context.Context, year int) (Report, error) {
    query := `
        SELECT 
            c.region,
            SUM(CASE WHEN i.status = 'paid' AND i.paid_date BETWEEN $1 AND $2 
                THEN i.amount ELSE 0 END) as total_paid,
            COUNT(i.id) as total_invoices
        FROM invoices i
        JOIN clients c ON i.client_id = c.id
        WHERE EXTRACT(year FROM i.created_at) = $3
        GROUP BY c.region
        ORDER BY total_paid DESC`

    // Explicit, optimized, debuggable
    rows, err := r.db.Query(ctx, query, startDate, endDate, year)
    // ... scanning logic
}

Musa smiled: "Perfect SQL, but managing the scanning is tedious."

Team Gamma's Multiplication

Sofia found herself copying similar complex queries across multiple handlers, each with slight variations. Maintaining consistency became a nightmare.

Act 3: The Production Crisis

A critical bug emerged: recently paid invoices were still showing as 'overdue' in the nightly batch job.

Team Alpha's Mystery

The bug was hidden in GORM's lifecycle hooks and implicit update behavior. Tracing the exact point where the status field was incorrectly handled required digging through GORM's internals.

"There's too much magic!" a frustrated developer exclaimed.

Team Beta's Clarity

Musa's team traced the issue to their InvoiceRepository.UpdateStatus method. The explicit SQL made debugging straightforward:

func (r *postgresInvoiceRepo) UpdateStatus(ctx context.Context, id int64, status string) error {
    query := `UPDATE invoices SET status = $1, updated_at = NOW() WHERE id = $2`
    _, err := r.db.Exec(ctx, query, status, id)
    return err
}

Team Gamma's Hunt

Sofia found multiple places where invoice statuses were updated, each with slightly different SQL. The lack of centralization made the bug hard to pin down and fix consistently.

The Scorecard

Aspect	Team Alpha (GORM)	Team Beta (Repository)	Team Gamma (Raw SQL)
Initial Speed	🚀 Very Fast	🛠️ Moderate	⚡ Instant
Complex Queries	😰 Struggled	😊 Controlled	😅 Repetitive
Debugging	🔍 Opaque	🎯 Clear	🕵️ Scattered
Maintainability	📉 Degraded	📈 Improved	📉 Chaotic
Testing	🧪 Database-dependent	✅ Mockable	🚫 Impossible

The Path Forward

Our story reveals that each approach has its place, but the Repository Pattern offers the best balance of control, maintainability, and testability for most Go applications.

However, Team Beta's experience highlights a remaining challenge: implementing repositories manually involves significant boilerplate—writing SQL strings, managing Scan() calls, and handling type mapping.

This is where sqlc enters the picture. It bridges the gap between the Repository Pattern's architectural benefits and the developer experience of working with raw SQL.

What Makes a Great Data Access Strategy?

Based on our analysis, an ideal approach should provide:

Clear separation of concerns ✅ Repository Pattern
Explicit, optimized SQL ✅ Raw SQL / sqlc
Type safety ✅ sqlc
Testability ✅ Repository Pattern
Minimal boilerplate ✅ sqlc
Performance transparency ✅ Raw SQL / sqlc

The Repository Pattern + sqlc combination delivers on all these requirements.

What's Next in This Series

In Part 2: Implementing Effective Repositories in Go & Confronting the Critiques, we'll dive into practical implementation:

Build a complete repository - First manually with pgx, then enhanced with sqlc
Demonstrate testing strategies - Mock repositories, integration tests, and best practices
Address common criticisms - "Over-engineering," "Just use an ORM," and other concerns
Show real-world patterns - Transaction management, caching, and performance optimization

Conclusion

The Repository Pattern isn't just another abstraction layer—it's a strategic architectural decision that pays dividends as your application grows. By decoupling business logic from data access concerns, you create code that's more testable, maintainable, and adaptable to changing requirements.

The choice isn't between raw SQL and ORMs—it's between chaos and clarity, between fighting your tools and working with them. The Repository Pattern, especially when combined with sqlc, offers the clarity and control that Go developers value while maintaining the performance characteristics that make Go shine in backend development.

Ready to see this pattern in action? Join me in Part 2, where I'll build a production-ready repository implementation that will transform how you think about data access in Go.

Are you currently struggling with data access patterns in your Go applications? What challenges have you faced with ORMs or raw SQL? Share your experiences in the comments below—your insights help the entire Go community learn and grow.

Automate the Gruntwork: Why Developers and Companies Should Use 'Patched'

Claret Ibeawuchi — Tue, 06 Aug 2024 12:26:29 +0000

Introduction
The Rise of Development Automation
Introducing Patched
Key Features of Patched
What Can Patched Do?
The Power of Customization
Comparison: Patched vs. Other Automation Tools
What Sets Patched Apart
The Impact of Patched on Team Productivity and Project Timelines
Here's a Quick Demo of the Core Features of Patched
Getting Started with Patched
Conclusion

TL;DR

Patched is a free, open-source software tool designed to automate repetitive development tasks, allowing developers to focus on coding and solving business problems. It features customizable workflows, integration with large language models, and a user-friendly interface. Patched enhances productivity by reducing time spent on code reviews, documentation, and security checks, leading to improved code quality and faster project timelines. Download Patched and start optimizing your development process today!

Introduction

As software developers, we often find ourselves caught in a paradox: we're hired to code, yet research shows that on average, developers only spend 52 minutes coding per day — about 4 hours and 21 minutes during a normal workweek from Monday to Friday. The majority of our time is consumed with distractions, disruptions, and meetings as well as system inefficiencies, such as slow reviews, slow builds, reading documentation, writing commit messages, reviewing pull requests, and bad tools.

All of which impact the first prerequisite for high-performing engineering teams: code time. Without time to focus and enter a state of flow, developers are limited in their ability to get work done. The best engineering organizations prioritize and defend code time by limiting meetings, reducing distractions, and removing constraints with the right tools and automations.

The Rise of Development Automation

To address this challenge, the software industry has embraced automation tools. Popular solutions like GitHub Actions and GitHub Copilot have become staples in many developers' toolkits. However, there's a new player in town that's set to change the game: Patched.

Introducing Patched

Patched is a free, open-source software tool designed to automate the grunt work in software development. It leverages AI workflows to handle tasks like PR reviews, bug fixing, and security patching, allowing developers to focus on what truly matters: solving business problems and writing innovative code.

Key Features of Patched

Self-hosted CLI agent or GUI interface
Integration with your preferred Large Language Models (LLMs)
Customizable workflows called "patchflows"

What Can Patched Do?

Patched offers a variety of built-in patchflows:

GenerateDocstring: Automatically documents undocumented code
GenerateReadme: Creates README files based on repository content
PrReview: Summarizes changes in pull requests
AutoFix: Scans and fixes security issues in branches
DependencyUpgrade: Automatically updates vulnerable versions of dependencies in your repository to the fixed version
ResolveIssue: Automatically resolves issues (or bugs) in your repository

The Power of Customization

One of Patched's strongest features is its flexibility. Users can modify existing patchflows or create new ones, either through code or using the intuitive GUI to adjust Directed Acyclic Graphs (DAGs).

Comparison: Patched vs. Other Automation Tools

While several tools exist to automate various aspects of software development, Patched stands out for its versatility and customizability. Let's compare Patched with some popular alternatives:

1. Patched vs. GitHub Copilot

GitHub Copilot: Focuses on AI-powered code completion and generation.
Patched: Offers a broader range of automation tasks beyond just code writing, including PR reviews, documentation generation, and security patching.
Key Advantage: Patched provides a more comprehensive solution for development workflow automation.

2. Patched vs. SonarQube

SonarQube: Specializes in code quality and security analysis.
Patched: While including code analysis features, Patched also automates other tasks like documentation and PR reviews.
Key Advantage: Patched offers a more diverse set of customizable workflows.

3. Patched vs. Jenkins

Jenkins: Primarily used for Continuous Integration and Continuous Delivery (CI/CD).
Patched: Focuses on automating development tasks rather than build and deployment processes.
Key Advantage: Patched addresses the day-to-day tasks of developers more directly.

What Sets Patched Apart

Open-Source Nature: Unlike many proprietary tools, Patched is free and open-source, allowing for community contributions and customizations.
Customizable Workflows: Patched's "patchflows" can be easily modified or created from scratch, adapting to specific team needs.
LLM Flexibility: Users can choose their preferred Large Language Models, offering more control over the AI assistance.
Unified Solution: Patched combines features that might otherwise require multiple separate tools, streamlining the development process.

The Impact of Patched on Team Productivity and Project Timelines

Implementing Patched in your development workflow can lead to significant improvements in team productivity and project timelines. Let's explore the potential impacts:

1. Reduced Time on Repetitive Tasks

Patched automates many time-consuming tasks such as code reviews, documentation generation, and security checks. This automation can save developers several hours each week, allowing them to focus on core development activities.

Example: A team of 10 developers each saving 5 hours per week could result in 50 additional hours of productive coding time - equivalent to gaining more than one full-time developer.

2. Faster Code Reviews and Merges

With Patched's PR review feature, pull requests can be initially reviewed by AI, highlighting potential issues before human review. This can speed up the review process and reduce the back-and-forth typically seen in PR discussions.

Potential Impact: Reduce average PR review time by 30-50%, allowing features to be merged and deployed more quickly.

3. Improved Code Quality

By consistently applying best practices in documentation, security checks, and code style, Patched helps maintain high code quality standards. This can lead to fewer bugs and reduced technical debt over time.

Long-term Benefit: Less time spent on bug fixes and refactoring, potentially reducing maintenance costs by 20-30%.

4. Accelerated Onboarding for New Team Members

With automatically generated and consistently maintained documentation, new team members can get up to speed more quickly on project codebases.

Estimate: Reduce onboarding time for new developers by 25-40%, allowing them to contribute meaningful code sooner.

5. Enhanced Security Practices

Regular automated security scans can catch vulnerabilities early in the development process, reducing the risk and cost associated with security issues found later in the project lifecycle.

Potential Saving: Avoid costly security breaches and reduce time spent on emergency patches by up to 60%.

6. Streamlined Project Timelines

By automating routine tasks and improving overall efficiency, Patched can help teams meet project deadlines more consistently and potentially even shorten development cycles.

Projected Improvement: Reduce overall project timelines by 15-25% for medium to large-scale projects.

Here's a Quick Demo of the Core Features of Patched

Getting Started with Patched

To get started with Patched, visit the official documentation website, join the Patched Discord community, and get started with building your first patchflow here. If you're confused, Patched offers an AI companion chatbot to help generate patchflows based on your prompts, which you can then fine-tune to your needs.

Conclusion

In an industry where time is money, Patched emerges as a powerful ally for developers and companies alike. By automating repetitive tasks, it allows teams to focus on innovation and problem-solving, ultimately leading to more efficient and effective software development. The potential for significant productivity gains and timeline improvements with Patched is substantial. As teams become more familiar with the tool and customize it to their specific needs, the benefits are likely to compound over time.

Whether you're a small startup or a large enterprise, Patched offers a versatile, customizable solution to streamline your development processes and boost your team's productivity. By embracing this open-source AI-powered tool, you're not just saving time – you're investing in your team's ability to innovate and deliver high-quality software faster than ever before.

Call to Action

Ready to revolutionize your development workflow? Download Patched today and join the growing community of developers who are already experiencing the benefits of this powerful tool. For more information and support, visit our documentation and join the Patched Discord community. Let's build the future of software development together!

Understanding Tests in Python: A Comprehensive Guide Using Pytest

Claret Ibeawuchi — Sun, 26 Nov 2023 17:59:13 +0000

What is a Test?
- 1.1 The Four Steps of Testing
Application Testing in Python with Pytest
- 2.1 Introduction to Pytest
- 2.2 Key Features of Pytest
- 2.3 Alternatives to Pytest
Conventions for Python Test Discovery
Getting Started with Pytest
- 4.1 Prerequisites
- 4.2 Installation
Writing Tests with Pytest
- 5.1 math_operations/math_operations.py
- 5.2 tests/conftest.py
- 5.3 tests/test_math_operations.py
- 5.4 pytest.ini
Understanding the Scripts
- 6.1 math_operations/math_operations.py
- 6.2 tests/conftest.py
- 6.3 tests/test_math_operations.py
- 6.4 pytest.ini
Important Concepts
- 7.1 Pytest Fixtures
- 7.2 Pytest Marks and Configurations
- 7.3 Command-line Flags You Should Know
Deployment Best Practice: Using Pytest with Tox
Conclusion

What is a Test?

Pause for a moment and ponder: What is a test? According to Google, a test is "a procedure intended to establish the quality, performance, or reliability of something, especially before it is taken into widespread use." In simpler terms, a test evaluates the outcome of a specific behavior to ensure it aligns with expectations.

Tests should be carried out before deployment to production.

The Four Steps of Testing

Arrange: Prepare the test environment, set the stage for the desired behavior. This involves tasks such as object preparation, service initialization, or data entry into a database.
Act: Trigger the singular, state-changing action that initiates the behavior to be tested. Typically, this involves calling a function or method.
Assert: Examine the resulting state to confirm it aligns with expectations. Assertions validate if the behavior meets the anticipated outcome. For example, assert result == "5".
Cleanup: Ensure the test environment returns to its initial state to prevent unintended influence on subsequent tests.

At its core, a test revolves around the Act and Assert steps, with Arrange providing the necessary context. The behavior under scrutiny unfolds between the Act and Assert phases.

Application Testing in Python with Pytest

Now, let's delve into application testing in Python using Pytest, a powerful, flexible, and extensible testing framework.

Introduction to Pytest

Pytest stands out as a preferred testing framework due to its simplicity and versatility. Its features include:

Easy to Use: Pytest boasts a straightforward and intuitive syntax, making test creation and execution enjoyable.
Flexible: Suited for testing a wide array of applications, Pytest supports various fixtures, assertions, and markers.
Extensible: Highly customizable with numerous third-party plugins, extending its functionality.

Key Features of Pytest

Improved Test Coverage: Easily write tests covering all facets of your application, enhancing overall code quality.
Reduced Regression Risk: Identify and rectify regressions early in development, saving time and effort.
Increased Developer Productivity: Streamlined test creation and execution free up developers to focus on other tasks.

Alternatives to Pytest

Unittest:
- Pros: Comes with the standard library, well-integrated with the Python ecosystem.
- Cons: Limited built-in fixtures, more boilerplate code compared to Pytest.
Nose:
- Pros: Extends unittest for easier testing, features a plugin architecture for extensibility.
- Cons: Less actively developed than Pytest, some feature overlap with unittest.

Conventions for Python Test Discovery

Before diving into Pytest, understanding its test discovery conventions is essential. Pytest follows a standard discovery process:

Collection starts from testpaths or the current directory.
Recursion into directories, excluding those listed in norecursedirs.
Searching for test_*.py or *_test.py files, imported by their test package name.
Collecting test items, including test functions and methods.

For optimal test discovery, name your pytest scripts as test_*.py or *_test.py for readability and consistency.

Getting Started with Pytest

Prerequisites

Ensure you have:

Basic Python knowledge
Understanding of Pip
Text Editor or IDE

Installation

Install Pytest using pip:

pip install pytest

Writing Tests with Pytest

Consider the following directory structure:

pytest-article/
│
├── math_operations/
│   ├── __init__.py
│   ├── math_operations.py
│
├── tests/
│   ├── conftest.py
│   ├── __init__.py
│   ├── test_math_operations.py
│
└── pytest.ini

`math_operations/math_operations.py`

# Code defining basic mathematical operations and a Calculator class
def add(x, y):
    return x + y

def subtract(x, y):
    return x - y

class Calculator:
    def multiply(self, x, y):
        return x * y

    def divide(self, x, y):
        if y == 0:
            raise ValueError("Cannot divide by zero")
        return x / y

`tests/conftest.py`

# Fixture to create an instance of the Calculator class for testing
import pytest
from math_operations.math_operations import Calculator

@pytest.fixture
def calculator_instance():
    return Calculator()

`tests/test_math_operations.py`


import pytest
from math_operations.math_operations import add, subtract

# Fixture to set up some common data for the tests
@pytest.fixture
def common_data():
    return {'x': 10, 'y': 5}

# Test the add function
 add function
def test_add(common_data):
    result = add(common_data['x'], common_data['y'])
    assert result == 15

# Test the subtract function
def test_subtract(common_data):
    result = subtract(common_data['x'], common_data['y'])
    assert result == 5

# Test the multiply method of the Calculator class
def test_calculator_multiply(calculator_instance, common_data):
    result = calculator_instance.multiply(common_data['x'], common_data['y'])
    assert result == 50

# Test the divide method of the Calculator class
def test_calculator_divide(calculator_instance, common_data):
    result = calculator_instance.divide(common_data['x'], common_data['y'])
    assert result == 2.0

# Test division by zero using a marker
@pytest.mark.divide_by_zero
def test_calculator_divide_by_zero(calculator_instance, common_data):
    with pytest.raises(ValueError, match="Cannot divide by zero"):
        calculator_instance.divide(common_data['x'], 0)

# Use marks to categorize tests
@pytest.mark.parametrize("input_x, input_y, expected_result", [(2, 3, 5), (0, 0, 0), (-2, -3, -5)])
def test_add_parametrized(input_x, input_y, expected_result):
    result = add(input_x, input_y)
    assert result == expected_result

`pytest.ini`

[pytest]
testpaths = tests # define test paths
markers =
    divide_by_zero: custom marker for tests that involve division by zero
addopts = -v # make output verbose

Run tests using:

pytest

Understanding the Scripts

math_operations/math_operations.py: Defines basic mathematical operations and a Calculator class.
tests/conftest.py: Provides a fixture, calculator_instance, creating a Calculator instance for testing.
tests/test_math_operations.py: Tests functions and methods from math_operations.py.
pytest.ini: Configuration file specifying test paths, markers, and additional options.

Important Concepts

Pytest Fixtures

Fixture Definition: conftest.py defines a fixture named calculator_instance, creating a shared Calculator instance for tests.
Fixture Sharing: Fixtures in conftest.py can be shared across test files in the same directory and subdirectories.
Organization and Reusability: Promotes code organization and reuse by centralizing setup logic, ensuring a consistent testing environment.

Pytest Marks and Configurations

Custom Mark: @pytest.mark.divide_by_zero categorizes the test_calculator_divide_by_zero function for division by zero scenarios.
Parametrize: @pytest.mark.parametrize streamlines testing with different inputs, reducing redundancy.
pytest.ini Configuration: Registers custom marks, sets testpaths, and includes additional options for test execution.

Command-line Flags You Should Know

-k EXPRESSION: Run tests matching the given substring expression.
--pdb: Start the interactive Python debugger on errors or KeyboardInterrupt.
--last-failed: Rerun only tests that failed in the last run.
-v, --verbose: Increase verbosity.
--disable-warnings, --disable-pytest-warnings: Disable warnings summary

Deployment Best Practice: Using Pytest with Tox

Consider integrating Tox, a virtualenv test automation tool, for comprehensive testing. Tox sets up virtual environments with predefined dependencies and executes configured test commands. This ensures tests run against the installed package, detecting potential packaging issues.

For a basic understanding of tox, check out my beginner’s guide article on tox

Also, here is a link to a repository where I implemented testing with pytest in a tox.ini file.rent predictor

Conclusion

In the vast landscape of software development, testing stands as a guardian, ensuring the quality, performance, and reliability of our creations. Armed with an understanding of tests, Pytest, and best practices, you're now equipped to elevate the quality and reliability of your Python applications through effective testing.

🌟 Share, Comment, and React! If you found this article helpful or need more clarity, your engagement is highly appreciated.

Happy coding! 🐍✨ Don't forget to test!!! 🧪

Happy coding! Don't forget to test!!!

Python Asyncio: A Guide to Asynchronous Programming and Concurrency

Claret Ibeawuchi — Wed, 15 Nov 2023 08:37:18 +0000

What is Concurrency?
Writing Concurrent Code in Python
Libraries for Asynchronous Programming
Getting Started with asyncio
- Asynchronous Programming in Action
  - Synchronous Code
  - Asynchronous Code
  - Walkthrough of Asynchronous Code
Most Popular asyncio Functions
Other Relevant Information
- Debug Mode
Conclusion

Python, known for its synchronous code execution, processes tasks line by line, causing delays when tasks take time to run. In a previous article, we discussed parallelization, the Global Interpreter Lock (GIL), and its impact on parallel execution.

To overcome this, we turn to concurrency.

What is Concurrency?

Concurrency involves running multiple tasks simultaneously by efficiently managing their wait times.

Writing Concurrent Code in Python

To achieve concurrency in Python, we turn to asynchronous programming, briefly discussed here.

Libraries for Asynchronous Programming

Three notable libraries for asynchronous programming are:

asyncio
anyio
aiohttp

This article focuses on asyncio, the foundation for anyio and aiohttp.

Getting Started with asyncio

asyncio enables writing concurrent code using the async/await syntax. It serves as the base for various Python asynchronous frameworks, offering high-performance solutions for network and web servers, database connections, distributed task queues, and more.

asyncio is particularly suitable for IO-bound and high-level structured network code. Its high-level APIs allow:

Running Python coroutines concurrently
Performing network IO and IPC
Controlling subprocesses
Distributing tasks via queues
Synchronizing concurrent code

Low-level APIs are also available for creating and managing event loops, implementing efficient protocols, and bridging callback-based libraries with async/await syntax.

Asynchronous Programming in Action

Synchronous Code

#sync.py
import time

def print_message(message, delay):
    time.sleep(delay)
    print(message)

def main():
    print_message("Hello", 2)
    print_message("Sync", 1)
    print_message("World", 3)

if __name__ == "__main__":
    start_time = time.time()
    main()
    end_time = time.time()
    print(f"Total execution time: {end_time - start_time} seconds")

Asynchronous Code

#async.py
import asyncio
import time

# Define a simple asynchronous function
async def print_message(message, delay):
    # Simulate some asynchronous operation, like I/O or network request
    await asyncio.sleep(delay)
    print(message)

# Define the main asynchronous function
async def main():
    # Use asyncio.gather to run multiple asynchronous functions concurrently
    # Wait for each task to complete in order
    await asyncio.gather(
        print_message("Async", 2),
        print_message("Hello", 1),
        print_message("World", 3)
    )

# Run the event loop to execute the asynchronous code
if __name__ == "__main__":
    # Record the start time for measuring execution time
    start_time = time.time()

    # Run the main asynchronous function using asyncio.run()
    asyncio.run(main())

    # Calculate and print the total execution time
    end_time = time.time()
    print(f"Total execution time: {end_time - start_time} seconds")

Walkthrough of Asynchronous Code:

Import the required libraries: asyncio and time.
Define an asynchronous function print_message(message, delay). This function simulates asynchronous operations using asyncio.sleep(delay) and prints the given message.
Define the main asynchronous function main(). It uses asyncio.gather to concurrently run multiple instances of the print_message function with different messages and delays.
Check if the script is being run directly using if __name__ == "__main__":.
Record the start time using start_time = time.time().
Run the main asynchronous function using asyncio.run(main()).
Record the end time using end_time = time.time().
Print the total execution time: print(f"Total execution time: {end_time - start_time} seconds").

Other Relevant Information

Debug Mode:
- Enable by setting PYTHONASYNCIODEBUG=1, using Python Development Mode, passing debug=True to asyncio.run(), or calling loop.set_debug().
- Configure the logger level to logging.DEBUG.
- Display ResourceWarning warnings using -W default command line option.

With debug mode:

asyncio checks for unawaited coroutines.
Non-threadsafe asyncio APIs raise exceptions if called from the wrong thread.
I/O selector execution time is logged if it takes too long.
Callbacks longer than 100 milliseconds are logged.

Conclusion

Incorporate asyncio into your Python projects for efficient concurrent and asynchronous programming. Explore its rich set of functions and keep in mind the debugging options for smoother development.

I will be writing an article on anyio. The asynchronous programming library used by fastapi and starlette to handle requests and other tasks.

In the mean time, follow me, jump to comments, and leabe positive reactions.

Happy asynchronous coding!!!

20 Keywords in Asynchronous Programming: Concurrency, Multiprocessing... all explained.

Claret Ibeawuchi — Wed, 15 Nov 2023 07:09:38 +0000

Welcome to the World of Asynchronous Programming in Python!

As developers, we often encounter scenarios where waiting for certain tasks, such as input/output operations, can lead to inefficiencies in our programs.

Quick one: Think of such situations, if any, comment, let's have a discussion

Asynchronous programming is needed in multiple situations such as: I/O-Bound Operations, Web Development, Concurrency in Networking, GUI Applications, Real-Time Applications, Long-Running Tasks.

In this article, we'll explore the shift from traditional synchronous programming to the asynchronous paradigm in Python, uncovering powerful concepts and tools that enhance the efficiency and responsiveness of our applications.

Here are brief explanations of a list of important keywords related to asynchronous programming in Python:

Synchronous (Classic Sequential) Programming

In synchronous programming, tasks run one after the other in a linear fashion. Each task must finish before the next one begins, leading to inefficiencies, especially in scenarios involving input/output operations.
Asynchronous Programming

Asynchronous code allows a program to signal that it will need to wait for certain tasks, such as input/output (I/O) operations, to complete. Instead of idly waiting, the program can continue executing other tasks. Common I/O-bound operations include network communication, file read/write, remote API calls, and database operations.

Concurrency, not Multiprocessing or Multithreading:
Asynchronous programming involves running one task at a time, not multithreading or multiprocessing. During the wait time of one function, another function can be executed.
Global Interpreter Lock (GIL)

The GIL in Python ensures that only one thread holds control of the interpreter at a time. While not impactful in single-threaded programs, it can be a bottleneck in CPU-bound and multi-threaded code.
CPython

CPython is the canonical implementation of the Python programming language, distributed on python.org. It's essential to distinguish CPython from other implementations like Jython or IronPython.
Process

A process is a separate instance of the Python interpreter, managed by the operating system and represented by a multiprocessing.Process object.
Thread

A thread is an entity within a process that can be scheduled for execution. Threads are smaller units of processing and are managed by the operating system.
Thread Pool

A thread pool is a collection of threads created in advance and reused to execute multiple tasks. Python's concurrent.futures.ThreadPoolExecutor simplifies thread pool management.
Threading

The concurrent.futures.ThreadPoolExecutor offers a higher-level interface for background thread execution. queue provides a thread-safe way to exchange data between threads, and asyncio offers task-level concurrency without the need for multiple OS threads.
Multithreading

Multithreading allows a processor to execute multiple threads concurrently, achieved through context switching.
Multiprocessing

Multiprocessing refers to a system's ability to support more than one processor simultaneously. Python's multiprocessing module leverages subprocesses, allowing full utilization of multiple processors.
Parallelisation

While modern computers can handle parallelism with multiple cores, Python's Global Interpreter Lock (GIL) limits true parallelism. The GIL restricts Python code to single-threaded execution.
Concurrency

Concurrency means running more than one task simultaneously, switching between tasks rather than running them in parallel.
Coroutine

Coroutines are a more generalized form of subroutines, executed using the async/await syntax. They allow the suspension and resumption of tasks, enhancing concurrency.
AsyncIO

AsyncIO is a library for writing concurrent code using async/await syntax. It serves as a foundation for various asynchronous frameworks in Python, suitable for IO-bound and high-level structured network code.
Awaitable

An awaitable is a Future-like object or a coroutine object, usable in an await expression.
Event Loop

An event loop runs in a thread and executes callbacks and tasks. Tasks are wrapped coroutines that can be executed independently.
Task

A task is a wrapped coroutine that can be executed independently.
Futures

Futures represent the result of an asynchronous computation.
AnyIO

AnyIO is an asynchronous networking and concurrency library compatible with both asyncio and trio. It implements structured concurrency on top of asyncio.
Aiohttp

Aiohttp is an asynchronous HTTP client/server for asyncio in Python.

Demystifying FastAPI's CORSMiddleware for Cross-Origin Requests

Claret Ibeawuchi — Sun, 05 Nov 2023 22:59:26 +0000

Content

Introduction
What is CORS?
- FastAPI's Definition
- Mozilla's Definition
- Technical Explanation
CORS Basics
- How CORS Works
- Handling Errors
Configuring FastAPI's CORSMiddleware
- Requirements
- Step 1: Install FastAPI
- Step 2: Necessary Imports
- Step 3: Create an Instance of the FastAPI Framework
- Step 4: Define Allowed Origins
- Step 5: Configure CORSMiddleware
- Step 6: Syntax for API Endpoint
- Explanation of Supported Configuration Arguments
Pre-Conclusion
Discovering Starlette Configuration and Code Underneath
Conclusion

Introduction

Cross-Origin Resource Sharing (CORS) is a critical mechanism that allows servers to control which origins (domains, schemes, or ports) can access their resources. In this article, we will delve into FastAPI's CORSMiddleware, understand its functionality, and learn how to configure it effectively.

What is CORS?

Think of CORS as a border protocol setup between the server side and the client side intended to allow resource hosts (any service that makes its data available via HTTP) to restrict which websites may access that data.

According to the FastAPI documentation, CORS or "Cross-Origin Resource Sharing" refers to the situations when a frontend running in a browser has JavaScript code that communicates with a backend, and the backend is in a different "origin" than the frontend.

According to Mozilla documentation, Cross-Origin Resource Sharing (CORS) is an HTTP-header based mechanism that allows a server to indicate any origins (domain, scheme, or port) other than its own from which a browser should permit loading resources.

In more technical terms, CORS, or Cross-Origin Resource Sharing, is an HTTP-header-based system that enables a server to specify which origins other than its own can access its resources, ensuring secure cross-origin data transfers while protecting against unauthorized requests.

CORS Basics

How CORS works

The Cross-Origin Resource Sharing standard works by adding new HTTP headers that let servers describe which origins are permitted to read that information from a web browser. Additionally, for HTTP request methods that can cause side-effects on server data (in particular, HTTP methods other than GET, or POST with certain MIME types), the specification mandates that browsers "preflight" the request, soliciting supported methods from the server with the HTTP OPTIONS request method, and then, upon "approval" from the server, sending the actual request. Servers can also inform clients whether "credentials" (such as Cookies and HTTP Authentication) should be sent with requests.

To achieve this, the backend must have a list of "allowed origins." In this case, it would have to include http://localhost:3000 or * for the frontend to work correctly.

Handling errors

CORS failures result in errors but, for security reasons, specifics about the error are not available to JavaScript. The only way to determine what specifically went wrong is to look at the browser's console for details. For a deeper understanding of CORS, visit the Mozilla documentation.

Configuring FastAPI's CORSMiddleware

FastAPI provides its own middleware for handling CORS, offering granular control over cross-origin requests. We'll dive into the specifics of configuring FastAPI's CORSMiddleware step-by-step.

Requirements:

Python 3
Updated pip (best practice)
Code editor
Basic knowledge of pip installation and using a code editor

Step 1 (Should be done in the terminal):

Install FastAPI

pip install fastapi[all]

Step 2 (All other steps from here should be done in the code editor environment):

Necessary imports

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

Step 3:

Create an instance of the FastAPI framework and assign it to a variable named app

app = FastAPI()

Step 4:

A list of origins that you plan to be permitted to make cross-origin requests

origins = [
    "http://localhost:3000",
]

Step 5:

Pass parameter configurations to app.add_middleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Step 6:

Syntax for API endpoint

@app.get("/")
async def main():
    return {"message": "Hello World"}

Here’s a link to a sample project on my github that utilizes this CORS configuration in the main.py file of the FASTAPI folder.

What the supported configuration arguments mean:

allow_origins: A list of origins that should be permitted to make cross-origin requests. For example, ['https://example.org', 'https://www.example.org']. You can use ['*'] to allow any origin.
allow_origin_regex: A regex string to match against origins that should be permitted to make cross-origin requests. For example, 'https://.*\.example\.org'.
allow_methods: A list of HTTP methods that should be allowed for cross-origin requests. Defaults to ['GET']. You can use ['*'] to allow all standard methods.
allow_headers: A list of HTTP request headers that should be supported for cross-origin requests. Defaults to an empty list. You can use ['*'] to allow all headers. The Accept, Accept-Language, Content-Language, and Content-Type headers are always allowed for CORS requests.
allow_credentials: Indicate that cookies should be supported for cross-origin requests. Defaults to False.
expose_headers: Indicate any response headers that should be made accessible to the browser. Defaults to an empty list.
max_age: Sets a maximum time in seconds for browsers to cache CORS responses. Defaults to 600.

Important: In our Demo we used "*", so let’s talk about it. "*" a "wildcard" is used to say that all are allowed. But, like the joker it can get tricky.

If wildcards are used for allow_origins, it will only allow certain types of communication, excluding everything that involves credentials: Cookies, Authorization headers like those used with Bearer Tokens, etc.

So, for everything to work correctly, it's better to specify explicitly the allowed origins.

Pre-Conclusion

FastAPI's CORSMiddleware empowers you to take control over cross-origin requests, securing your APIs while allowing legitimate access. With the knowledge gained from this article, you can confidently configure and utilize this middleware for your FastAPI projects, ensuring secure and efficient cross-origin data transfers.

Start building!

If you’re a nerd like me and curious about what code is down imports deeper in FastAPI CORSMiddleware, then it’s inception time.

Discovering Starlette Configuration and Code Underneath

So, I kept pressing ‘Go to definition’...

...

…and then I came across the ASGI gem Starlette

So, to keep the article length optimal, I’ll be writing about it in my next article. Keep checking my page, comment and like my posts.

Conclusion

Since this is not exactly where it ends (await the post conclusion :)), the conclusion will be in my next article.

Thank you for reading up to this point. Did you enjoy this article? Leave comments, questions, or other suggestions. Follow my page, contact me for technical articles, or collaborations on my LinkedIn.

DEV Community: Claret Ibeawuchi

Callback the Police: Enforcing Business Rules in AI Agents 👮‍♂️

Table of Contents

The Traffic Cop Analogy

What Are Callbacks in AI Agents?

The Callback Lifecycle

The before_tool_callback: Your First Line of Defense

Signature and Components

What You Can Do

Real-World Pattern: Admin Payment Exemption

Implementation

What Happens

Key Insights

Performance Optimization: Caching Expensive Checks

Real-World Pattern: Tool Interception and Routing

Implementation

What Happens

Benefits

Real-World Pattern: Auto-Injecting Context

Implementation

Storing Quote Data

User Experience

Key Insights

Common Mistakes and How to Avoid Them

❌ Mistake 1: Not Returning Anything

❌ Mistake 2: Modifying Args Without In-Place Updates

❌ Mistake 3: Forgetting Tool Context Parameter

❌ Mistake 4: Missing Tool Name Variants

❌ Mistake 5: Catching All Tools Too Broadly

❌ Mistake 6: No Logging

Real-World Pattern: Detecting Admin Message Leaks

The Problem

The Solution: Response Content Validation

Why This Happens

When to Use This Pattern

When to Use Which Callback

before_tool_callback - Pre-Execution Enforcement

after_model_callback - Post-Generation Validation

Comparison

Key Takeaways

The Enforcement Hierarchy

Resources

How Reliable Are Your AI Agents?

Table of Contents

The $117 Bug That Made Me Rethink Architecture

The Reliability Gap

What LLMs Are Good At

What LLMs Struggle With

The Developer's Dilemma

Why LLMs Fail at Critical Tasks

1. Token-Based Processing

2. Probabilistic, Not Deterministic

3. Context Window Limitations

4. No Inherent Security Model

5. Instruction Drift

The Multi-Layer Defense Strategy

The Architecture

Example: Catching the $117 Bug

Real-World Impact

Before: Hope and Pray 🙏

After: Defense in Depth 🛡️

The Reliability Equation

What's Next

Article 2: "Callback the Police" 👮‍♂️

Article 3: "The Ground Truth Principle" 📊

Article 4: "Explicit Contracts Save Lives" ⚖️

Article 5: "Event-Driven Automation" 🔄

Try It Yourself

Key Takeaways

Resources

Understanding Google Maps Grounding with ADK (Part 2/5)

Introduction

The Challenge: Limited Documentation

What You'll Learn

Google Maps Grounding Quickstart

Prerequisites

1. Set up Environment & Install ADK

2. Enable Vertex AI and Authenticate

3. Create Agent Project

4. Configure Environment Variables

`before_tool_callback` - Pre-Execution Enforcement

`after_model_callback` - Post-Generation Validation

5. Edit `agent.py`

Issue 1: `ValueError: google_maps parameter is not supported`

Issue 2: `DefaultCredentialsError`