async def get_valid_credentials(
      self, tool_context: ToolContext
  ) -> Optional[Credentials]:
    """Get valid credentials, handling refresh and OAuth flow as needed.

    Args:
        tool_context: The tool context for OAuth flow and state management

    Returns:
        Valid Credentials object, or None if OAuth flow is needed
    """
    # First, try to get credentials from the tool context
    creds_json = tool_context.state.get(BIGQUERY_TOKEN_CACHE_KEY, None)
    creds = (
        Credentials.from_authorized_user_info(
            json.loads(creds_json), self.credentials_config.scopes
        )
        if creds_json
        else None
    )

    # If credentails are empty use the default credential
    if not creds:
      creds = self.credentials_config.credentials

    # Check if we have valid credentials
    if creds and creds.valid:
      return creds

    # Try to refresh expired credentials
    if creds and creds.expired and creds.refresh_token:
      try:
        creds.refresh(Request())
        if creds.valid:
          # Cache the refreshed credentials
          tool_context.state[BIGQUERY_TOKEN_CACHE_KEY] = creds.to_json()
          return creds
      except RefreshError:
        # Refresh failed, need to re-authenticate
        pass

    # Need to perform OAuth flow
    return await self._perform_oauth_flow(tool_context)
Get valid credentials, handling refresh and OAuth flow as needed.

    Args:
        tool_context: The tool context for OAuth flow and state management

    Returns:
        Valid Credentials object, or None if OAuth flow is needed
    
src/google/adk/tools/bigquery/bigquery_credentials.py