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

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

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) 

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) 

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) 

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) 

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 

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}") 

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 

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 

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 

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" 

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 

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 

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() } 

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)}" 

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 

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}" 

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) 

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) 

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 ] 

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]