diff --git a/pkg/auth/direct_sftp_asset.go b/pkg/auth/direct_sftp_asset.go
new file mode 100644
index 00000000..5be5faa2
--- /dev/null
+++ b/pkg/auth/direct_sftp_asset.go
@@ -0,0 +1,326 @@
+package auth
+
+import (
+	"errors"
+	"strings"
+	"sync"
+
+	"github.com/gliderlabs/ssh"
+	gossh "golang.org/x/crypto/ssh"
+
+	"github.com/jumpserver-dev/sdk-go/model"
+	"github.com/jumpserver-dev/sdk-go/service"
+	"github.com/jumpserver/koko/pkg/config"
+	"github.com/jumpserver/koko/pkg/logger"
+	"github.com/jumpserver/koko/pkg/srvconn"
+)
+
+type AuthPhase string
+
+const (
+	AuthPhaseNone                       AuthPhase = ""
+	AuthPhaseUserKeyboardInteractive    AuthPhase = "user_keyboard_interactive"
+	AuthPhaseDirectSFTPAssetInteractive AuthPhase = "direct_sftp_asset_keyboard_interactive"
+)
+
+type directSFTPAssetAuthState struct {
+	Token     *model.ConnectToken
+	ctxDone   <-chan struct{}
+	sessionID string
+
+	mu            sync.Mutex
+	currentPrompt *directSFTPAssetPrompt
+	events        chan directSFTPAssetAuthEvent
+}
+
+type directSFTPAssetAuthEvent struct {
+	prompt   *directSFTPAssetPrompt
+	prepared *srvconn.PreparedDirectSFTP
+	err      error
+}
+
+type directSFTPAssetPrompt struct {
+	User        string
+	Instruction string
+	Questions   []string
+	Echos       []bool
+
+	answerCh chan directSFTPAssetPromptAnswer
+}
+
+type directSFTPAssetPromptAnswer struct {
+	answers []string
+	err     error
+}
+
+func completeSSHAuth(ctx ssh.Context) error {
+	ctx.SetValue(ContextKeyAuthPhase, AuthPhaseNone)
+	return prepareDirectSFTPAssetIfNeeded(ctx)
+}
+
+func continueDirectSFTPAssetAuth(ctx ssh.Context, challenger gossh.KeyboardInteractiveChallenge) error {
+	state, ok := ctx.Value(ContextKeyDirectSFTPAssetAuthState).(*directSFTPAssetAuthState)
+	if !ok || state == nil || state.Token == nil {
+		logger.Errorf("SSH conn[%s] direct sftp asset auth state not found", ctx.SessionID())
+		return authErr
+	}
+	for {
+		prompt := state.currentPendingPrompt()
+		if prompt == nil {
+			event := state.waitForNextEvent()
+			switch {
+			case event.prompt != nil:
+				prompt = event.prompt
+			case event.prepared != nil:
+				storePreparedDirectSFTP(ctx, event.prepared)
+				ctx.SetValue(ContextKeyDirectSFTPAssetAuthState, nil)
+				ctx.SetValue(ContextKeyAuthPhase, AuthPhaseNone)
+				return nil
+			default:
+				logger.Errorf("SSH conn[%s] direct sftp asset keyboard auth failed: %s",
+					ctx.SessionID(), event.err)
+				return authErr
+			}
+		}
+
+		answers, err := challenger(prompt.User, prompt.Instruction, prompt.Questions, prompt.Echos)
+		if submitErr := state.submitPromptAnswers(answers, err); submitErr != nil {
+			logger.Errorf("SSH conn[%s] direct sftp submit client answers failed: %s", ctx.SessionID(), submitErr)
+			return authErr
+		}
+		if err != nil {
+			logger.Errorf("SSH conn[%s] direct sftp interactive client answer failed: %s", ctx.SessionID(), err)
+			return authErr
+		}
+	}
+}
+
+func prepareDirectSFTPAssetIfNeeded(ctx ssh.Context) error {
+	req, ok := ctx.Value(ContextKeyDirectLoginFormat).(*DirectLoginAssetReq)
+	if !ok || req == nil {
+		return nil
+	}
+	if req.Protocol != model.ProtocolSFTP {
+		return nil
+	}
+	if prepared, ok := ctx.Value(ContextKeyPreparedDirectSFTP).(*srvconn.PreparedDirectSFTP); ok && prepared != nil && prepared.IsValid() {
+		return nil
+	}
+	user, ok := ctx.Value(ContextKeyUser).(*model.User)
+	if !ok || user == nil || user.ID == "" {
+		logger.Errorf("SSH conn[%s] direct sftp auth user not found", ctx.SessionID())
+		return authErr
+	}
+	jmsService, ok := ctx.Value(ContextKeyJMService).(*service.JMService)
+	if !ok || jmsService == nil {
+		logger.Errorf("SSH conn[%s] direct sftp auth jms service not found", ctx.SessionID())
+		return authErr
+	}
+	token, err := BuildDirectConnectToken(ctx, jmsService, user, req)
+	if err != nil {
+		logger.Errorf("SSH conn[%s] build direct sftp token failed: %s",
+			ctx.SessionID(), err)
+		return authErr
+	}
+	state := newDirectSFTPAssetAuthState(ctx.SessionID(), token, ctx.Done())
+	state.startHandshake()
+	event := state.waitForNextEvent()
+	switch {
+	case event.prepared != nil:
+		storePreparedDirectSFTP(ctx, event.prepared)
+		ctx.SetValue(ContextKeyAuthPhase, AuthPhaseNone)
+		return nil
+	case event.prompt != nil:
+		logger.Infof("SSH conn[%s] direct sftp prepare requires keyboard-interactive", ctx.SessionID())
+		ctx.SetValue(ContextKeyDirectSFTPAssetAuthState, state)
+		ctx.SetValue(ContextKeyAuthPhase, AuthPhaseDirectSFTPAssetInteractive)
+		return &ssh.PartialSuccessError{Next: ssh.ServerAuthCallbacks{
+			KeyboardInteractiveCallback: SSHKeyboardInteractiveAuth,
+		}}
+	default:
+		logger.Errorf("SSH conn[%s] prepare direct sftp asset failed: %s",
+			ctx.SessionID(), event.err)
+		return authErr
+	}
+}
+
+func buildDirectSFTPChallengeBridge(sessionID string, token *model.ConnectToken,
+	challenger gossh.KeyboardInteractiveChallenge) gossh.KeyboardInteractiveChallenge {
+	password := ""
+	if token != nil && !token.Account.IsSSHKey() {
+		password = token.Account.Secret
+	}
+	displayUser := ""
+	if token != nil {
+		displayUser = token.Account.Username
+	}
+	return func(user, instruction string, questions []string, echos []bool) (answers []string, err error) {
+		if len(questions) == 0 {
+			return []string{}, nil
+		}
+
+		answers = make([]string, len(questions))
+		pendingQuestions := make([]string, 0, len(questions))
+		pendingEchos := make([]bool, 0, len(questions))
+		pendingIndexes := make([]int, 0, len(questions))
+		for i := range questions {
+			if password != "" && isPasswordQuestion(questions[i]) {
+				answers[i] = password
+				continue
+			}
+			echo := false
+			if i < len(echos) {
+				echo = echos[i]
+			}
+			pendingIndexes = append(pendingIndexes, i)
+			pendingQuestions = append(pendingQuestions, questions[i])
+			pendingEchos = append(pendingEchos, echo)
+		}
+		autoFilledCount := len(questions) - len(pendingIndexes)
+		if len(pendingIndexes) == 0 {
+			return answers, nil
+		}
+
+		challengeUser := user
+		if challengeUser == "" {
+			challengeUser = displayUser
+		}
+		logger.Infof("SSH conn[%s] direct sftp interactive forwarding %d/%d questions to client (auto-filled %d)",
+			sessionID, len(pendingIndexes), len(questions), autoFilledCount)
+		pendingAnswers, err := challenger(challengeUser, instruction, pendingQuestions, pendingEchos)
+		if err != nil {
+			logger.Errorf("SSH conn[%s] direct sftp interactive client challenge failed: %s",
+				sessionID, err)
+			return nil, err
+		}
+		if len(pendingAnswers) != len(pendingIndexes) {
+			return nil, errors.New("direct sftp challenge answers mismatch")
+		}
+		for i, answer := range pendingAnswers {
+			answers[pendingIndexes[i]] = answer
+		}
+		return answers, nil
+	}
+}
+
+func newDirectSFTPAssetAuthState(sessionID string, token *model.ConnectToken, ctxDone <-chan struct{}) *directSFTPAssetAuthState {
+	return &directSFTPAssetAuthState{
+		Token:     token,
+		ctxDone:   ctxDone,
+		sessionID: sessionID,
+		events:    make(chan directSFTPAssetAuthEvent, 1),
+	}
+}
+
+func (s *directSFTPAssetAuthState) startHandshake() {
+	go func() {
+		prepared, err := newPreparedDirectSFTP(s.Token, srvconn.SSHClientKeyboardAuth(
+			buildDirectSFTPChallengeBridge(s.sessionID, s.Token, s.awaitClientAnswers),
+		))
+		select {
+		case s.events <- directSFTPAssetAuthEvent{prepared: prepared, err: err}:
+		case <-s.ctxDone:
+			if prepared != nil && prepared.Client != nil {
+				_ = prepared.Client.Close()
+			}
+		}
+	}()
+}
+
+func (s *directSFTPAssetAuthState) awaitClientAnswers(user, instruction string, questions []string, echos []bool) ([]string, error) {
+	prompt := &directSFTPAssetPrompt{
+		User:        user,
+		Instruction: instruction,
+		Questions:   append([]string(nil), questions...),
+		Echos:       append([]bool(nil), echos...),
+		answerCh:    make(chan directSFTPAssetPromptAnswer, 1),
+	}
+	s.setCurrentPendingPrompt(prompt)
+	select {
+	case s.events <- directSFTPAssetAuthEvent{prompt: prompt}:
+	case <-s.ctxDone:
+		s.setCurrentPendingPrompt(nil)
+		return nil, errors.New("direct sftp auth canceled")
+	}
+
+	select {
+	case answer := <-prompt.answerCh:
+		s.setCurrentPendingPrompt(nil)
+		return answer.answers, answer.err
+	case <-s.ctxDone:
+		s.setCurrentPendingPrompt(nil)
+		return nil, errors.New("direct sftp auth canceled")
+	}
+}
+
+func (s *directSFTPAssetAuthState) waitForNextEvent() directSFTPAssetAuthEvent {
+	select {
+	case event := <-s.events:
+		return event
+	case <-s.ctxDone:
+		return directSFTPAssetAuthEvent{err: errors.New("direct sftp auth canceled")}
+	}
+}
+
+func (s *directSFTPAssetAuthState) currentPendingPrompt() *directSFTPAssetPrompt {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.currentPrompt
+}
+
+func (s *directSFTPAssetAuthState) setCurrentPendingPrompt(prompt *directSFTPAssetPrompt) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.currentPrompt = prompt
+}
+
+func (s *directSFTPAssetAuthState) takeCurrentPendingPrompt() *directSFTPAssetPrompt {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	prompt := s.currentPrompt
+	s.currentPrompt = nil
+	return prompt
+}
+
+func (s *directSFTPAssetAuthState) submitPromptAnswers(answers []string, err error) error {
+	prompt := s.takeCurrentPendingPrompt()
+	if prompt == nil {
+		return errors.New("direct sftp prompt not found")
+	}
+	select {
+	case prompt.answerCh <- directSFTPAssetPromptAnswer{answers: answers, err: err}:
+		return nil
+	case <-s.ctxDone:
+		return errors.New("direct sftp auth canceled")
+	}
+}
+
+func isPasswordQuestion(question string) bool {
+	normalized := strings.ToLower(strings.TrimSpace(question))
+	return strings.Contains(normalized, "password")
+}
+
+func newPreparedDirectSFTP(token *model.ConnectToken, extraOpts ...srvconn.SSHClientOption) (*srvconn.PreparedDirectSFTP, error) {
+	sshClient, err := srvconn.NewSSHClientWithToken(token, config.GetConf().SSHTimeout, extraOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return &srvconn.PreparedDirectSFTP{
+		Token:              token,
+		Client:             sshClient,
+		DisableIdleRecycle: true,
+	}, nil
+}
+
+func storePreparedDirectSFTP(ctx ssh.Context, prepared *srvconn.PreparedDirectSFTP) {
+	ctx.SetValue(ContextKeyPreparedDirectSFTP, prepared)
+	go func(done <-chan struct{}, client *srvconn.SSHClient) {
+		<-done
+		_ = client.Close()
+	}(ctx.Done(), prepared.Client)
+}
+
+func isPartialSuccess(err error) bool {
+	var partialErr *ssh.PartialSuccessError
+	return errors.As(err, &partialErr)
+}
diff --git a/pkg/auth/direct_token.go b/pkg/auth/direct_token.go
new file mode 100644
index 00000000..8891bd17
--- /dev/null
+++ b/pkg/auth/direct_token.go
@@ -0,0 +1,122 @@
+package auth
+
+import (
+	"errors"
+	"fmt"
+	"net"
+
+	"github.com/gliderlabs/ssh"
+
+	"github.com/jumpserver-dev/sdk-go/common"
+	"github.com/jumpserver-dev/sdk-go/model"
+	"github.com/jumpserver-dev/sdk-go/service"
+	"github.com/jumpserver/koko/pkg/i18n"
+	"github.com/jumpserver/koko/pkg/logger"
+)
+
+func GetMatchedAssetsByDirectReq(jmsService *service.JMService, user *model.User, req *DirectLoginAssetReq) ([]model.PermAsset, error) {
+	var getUserPermAssets func() ([]model.PermAsset, error)
+	if common.IsUUID(req.AssetTarget) {
+		getUserPermAssets = func() ([]model.PermAsset, error) {
+			return jmsService.GetUserPermAssetsById(user.ID, req.AssetTarget)
+		}
+	} else {
+		getUserPermAssets = func() ([]model.PermAsset, error) {
+			return jmsService.GetUserPermAssetsByIP(user.ID, req.AssetTarget)
+		}
+	}
+	i18nLang := i18n.NewLang(user.Language)
+	assets, err := getUserPermAssets()
+	if err != nil {
+		logger.Errorf("Get user %s perm asset failed: %s", user.String(), err)
+		return nil, fmt.Errorf("match asset failed: %s", i18nLang.T("Core API failed"))
+	}
+	if len(assets) == 0 {
+		logger.Infof("User %s no perm for asset %s", user.String(), req.AssetTarget)
+		return nil, fmt.Errorf("match asset failed: %s", i18nLang.T("No found asset"))
+	}
+	return assets, nil
+}
+
+func MatchAccountsByUsername(accounts []model.PermAccount, username string) []model.PermAccount {
+	matched := make([]model.PermAccount, 0, len(accounts))
+	for i := range accounts {
+		account := accounts[i]
+		if account.Username == username {
+			matched = append(matched, account)
+		}
+	}
+	return matched
+}
+
+func GetMatchedAccounts(req *DirectLoginAssetReq, permAssetDetail model.PermAssetDetail) ([]model.PermAccount, error) {
+	return MatchAccountsByUsername(permAssetDetail.PermedAccounts, req.AccountUsername), nil
+}
+
+func BuildDirectConnectToken(ctx ssh.Context, jmsService *service.JMService, user *model.User,
+	req *DirectLoginAssetReq) (*model.ConnectToken, error) {
+	if req.IsToken() {
+		return req.ConnectToken, nil
+	}
+	selectedAssets, err := GetMatchedAssetsByDirectReq(jmsService, user, req)
+	if err != nil {
+		return nil, err
+	}
+	i18nLang := i18n.NewLang(user.Language)
+	if len(selectedAssets) != 1 {
+		msg := fmt.Sprintf(i18nLang.T("Must be unique asset for %s"), req.AssetTarget)
+		return nil, errors.New(msg)
+	}
+	permAssetDetail, err := jmsService.GetUserPermAssetDetailById(user.ID, selectedAssets[0].ID)
+	if err != nil {
+		msg := fmt.Sprintf(i18nLang.T("Must be unique asset for %s"), req.AssetTarget)
+		logger.Errorf("Get permAssetDetail failed: %s", err)
+		return nil, errors.New(msg)
+	}
+	if !permAssetDetail.SupportProtocol(req.Protocol) {
+		msg := fmt.Sprintf("not %s asset connection", req.Protocol)
+		logger.Errorf("Direct Request %s failed: %s", req.Protocol, msg)
+		return nil, errors.New(msg)
+	}
+
+	selectAccounts, err := GetMatchedAccounts(req, permAssetDetail)
+	if err != nil {
+		return nil, err
+	}
+	if len(selectAccounts) != 1 {
+		msg := fmt.Sprintf(i18nLang.T("Must be unique account for %s"), req.AccountUsername)
+		logger.Error(msg)
+		return nil, errors.New(msg)
+	}
+	selectAccount := selectAccounts[0]
+	switch selectAccount.Username {
+	case model.InputUser, model.DynamicUser, model.ANONUser:
+		msg := fmt.Sprintf(i18nLang.T("Must be auto login account for %s"), req.AccountUsername)
+		logger.Error(msg)
+		return nil, errors.New(msg)
+	}
+	remoteAddr, _, _ := net.SplitHostPort(ctx.RemoteAddr().String())
+	sessReq := &service.SuperConnectTokenReq{
+		UserId:        user.ID,
+		AssetId:       permAssetDetail.ID,
+		Account:       selectAccount.Alias,
+		Protocol:      req.Protocol,
+		ConnectMethod: req.Protocol,
+		RemoteAddr:    remoteAddr,
+	}
+	tokenInfo, err := jmsService.CreateSuperConnectToken(sessReq)
+	if err != nil {
+		msg := err.Error()
+		if tokenInfo.Detail != "" {
+			msg = tokenInfo.Detail
+		}
+		logger.Errorf("Create super connect token failed: %s", msg)
+		return nil, err
+	}
+	connectToken, err := jmsService.GetConnectTokenInfo(tokenInfo.ID, true)
+	if err != nil {
+		logger.Errorf("Create super connect token err: %s", err)
+		return nil, err
+	}
+	return &connectToken, nil
+}
diff --git a/pkg/auth/ssh.go b/pkg/auth/ssh.go
index a1b60a6c..59397c9b 100644
--- a/pkg/auth/ssh.go
+++ b/pkg/auth/ssh.go
@@ -32,19 +32,31 @@ func SSHPasswordAndPublicKeyAuth(jmsService *service.JMService) SSHAuthFunc {
 		},
 	}}
 	return func(ctx ssh.Context, password, publicKey string) error {
+		ctx.SetValue(ContextKeyJMService, jmsService)
 		if password == "" && publicKey == "" {
 			logger.Errorf("SSH conn[%s] no password and publickey", ctx.SessionID())
 			return authErr
 		}
 		remoteAddr, _, _ := net.SplitHostPort(ctx.RemoteAddr().String())
 		username := ctx.User()
+		action := actionAccepted
+		var res error
 		if req, ok := parseDirectLoginReq(jmsService, ctx); ok {
 			if req.IsToken() {
 				if req.Authenticate(password) {
 					ctx.SetValue(ContextKeyUser, &req.ConnectToken.User)
+					res = completeSSHAuth(ctx)
+					switch {
+					case res == nil:
+						action = actionAccepted
+					case isPartialSuccess(res):
+						action = actionPartialAccepted
+					default:
+						action = actionFailed
+					}
 					logger.Infof("SSH conn[%s] %s for %s from %s", ctx.SessionID(),
-						actionAccepted, username, remoteAddr)
-					return nil
+						action, username, remoteAddr)
+					return res
 				} else {
 					logger.Errorf("SSH conn[%s] token %s auth failed", ctx.SessionID(), req.ConnectToken.Id)
 					return authErr
@@ -53,8 +65,6 @@ func SSHPasswordAndPublicKeyAuth(jmsService *service.JMService) SSHAuthFunc {
 			username = req.User()
 		}
 		authMethod := "publickey"
-		action := actionAccepted
-		var res error
 		if password != "" {
 			authMethod = "password"
 		}
@@ -114,10 +124,20 @@ func SSHPasswordAndPublicKeyAuth(jmsService *service.JMService) SSHAuthFunc {
 				res = needPasswordAuthErr
 				ctx.SetValue(ContextKeyCurrentAuth, "password")
 				action = actionPartialAccepted
+			} else {
+				res = completeSSHAuth(ctx)
+				switch {
+				case res == nil:
+				case isPartialSuccess(res):
+					action = actionPartialAccepted
+				default:
+					action = actionFailed
+				}
 			}
 		case authConfirmRequired, authMFARequired:
 			action = actionPartialAccepted
 			ctx.SetValue(ContextKeyAuthStatus, authStatus)
+			ctx.SetValue(ContextKeyAuthPhase, AuthPhaseUserKeyboardInteractive)
 			res = &ssh.PartialSuccessError{Next: ssh.ServerAuthCallbacks{
 				KeyboardInteractiveCallback: SSHKeyboardInteractiveAuth,
 			}}
@@ -142,6 +162,9 @@ func SSHKeyboardInteractiveAuth(ctx ssh.Context, challenger gossh.KeyboardIntera
 	if value, ok := ctx.Value(ContextKeyAuthFailed).(*bool); ok && *value {
 		return authErr
 	}
+	if phase, ok := ctx.Value(ContextKeyAuthPhase).(AuthPhase); ok && phase == AuthPhaseDirectSFTPAssetInteractive {
+		return continueDirectSFTPAssetAuth(ctx, challenger)
+	}
 
 	username := GetUsernameFromSSHCtx(ctx)
 	client, ok := ctx.Value(ContextKeyClient).(*UserAuthClient)
@@ -165,7 +188,7 @@ func SSHKeyboardInteractiveAuth(ctx ssh.Context, challenger gossh.KeyboardIntera
 		return authErr
 	}
 	if checkAuth != nil && checkAuth(ctx, challenger) {
-		return nil
+		return completeSSHAuth(ctx)
 	}
 	return authErr
 }
@@ -185,6 +208,14 @@ const (
 	ContextKeyCurrentAuth = "CONTEXT_CURRENT_AUTH"
 
 	ContextKeyAuthCount = "CONTEXT_AUTH_COUNT"
+
+	ContextKeyJMService = "CONTEXT_JMS_SERVICE"
+
+	ContextKeyAuthPhase = "CONTEXT_AUTH_PHASE"
+
+	ContextKeyPreparedDirectSFTP = "CONTEXT_PREPARED_DIRECT_SFTP"
+
+	ContextKeyDirectSFTPAssetAuthState = "CONTEXT_DIRECT_SFTP_ASSET_AUTH_STATE"
 )
 
 type DirectLoginAssetReq struct {
diff --git a/pkg/handler/direct_handler.go b/pkg/handler/direct_handler.go
index bce2ff99..c97fb293 100644
--- a/pkg/handler/direct_handler.go
+++ b/pkg/handler/direct_handler.go
@@ -13,6 +13,7 @@ import (
 
 	"github.com/jumpserver-dev/sdk-go/model"
 	"github.com/jumpserver-dev/sdk-go/service"
+	"github.com/jumpserver/koko/pkg/auth"
 	"github.com/jumpserver/koko/pkg/common"
 	"github.com/jumpserver/koko/pkg/i18n"
 	"github.com/jumpserver/koko/pkg/logger"
@@ -53,7 +54,8 @@ type directOpt struct {
 	User          *model.User
 	terminalConf  *model.TerminalConfig
 
-	tokenInfo *model.ConnectToken
+	tokenInfo          *model.ConnectToken
+	preparedDirectSFTP *srvconn.PreparedDirectSFTP
 
 	sftpMode bool
 
@@ -100,6 +102,12 @@ func DirectConnectToken(tokenInfo *model.ConnectToken) DirectOpt {
 	}
 }
 
+func DirectPreparedSFTP(prepared *srvconn.PreparedDirectSFTP) DirectOpt {
+	return func(opts *directOpt) {
+		opts.preparedDirectSFTP = prepared
+	}
+}
+
 func DirectConnectProtocol(protocol string) DirectOpt {
 	return func(opts *directOpt) {
 		opts.protocol = protocol
@@ -165,7 +173,7 @@ func (d *DirectHandler) NewSFTPHandler() *SftpHandler {
 		if len(d.opts.assets) == 1 {
 			asset := d.opts.assets[0]
 			if permAssetDetail, err := d.jmsService.GetUserPermAssetDetailById(d.opts.User.ID, asset.ID); err == nil {
-				matchedAccount := GetMatchedAccounts(permAssetDetail.PermedAccounts, d.opts.targetAccount)
+				matchedAccount := auth.MatchAccountsByUsername(permAssetDetail.PermedAccounts, d.opts.targetAccount)
 				if len(matchedAccount) == 1 {
 					selectAccount := &matchedAccount[0]
 					req := service.SuperConnectTokenReq{
@@ -188,6 +196,9 @@ func (d *DirectHandler) NewSFTPHandler() *SftpHandler {
 		opts = append(opts, srvconn.WithAccountUsername(d.opts.targetAccount))
 	} else {
 		opts = append(opts, srvconn.WithConnectToken(d.opts.tokenInfo))
+		if d.opts.preparedDirectSFTP != nil && d.opts.preparedDirectSFTP.IsValid() {
+			opts = append(opts, srvconn.WithPreparedDirectSFTP(d.opts.preparedDirectSFTP))
+		}
 	}
 	return &SftpHandler{
 		UserSftpConn: srvconn.NewUserSftpConn(d.jmsService, opts...),
@@ -395,7 +406,7 @@ func (d *DirectHandler) Proxy(asset model.PermAsset) {
 		utils.IgnoreErrWriteString(d.term, lang.T("Core API failed"))
 		return
 	}
-	matched := GetMatchedAccounts(permAssetDetail.PermedAccounts, d.opts.targetAccount)
+	matched := auth.MatchAccountsByUsername(permAssetDetail.PermedAccounts, d.opts.targetAccount)
 	if len(matched) == 0 {
 		msg := fmt.Sprintf(lang.T("not found matched username %s"), d.opts.targetAccount)
 		utils.IgnoreErrWriteString(d.term, msg+"\r\n")
@@ -478,14 +489,3 @@ func (d *DirectHandler) Proxy(asset model.PermAsset) {
 	}
 	d.LoginConnectToken(&connectToken)
 }
-
-func GetMatchedAccounts(accounts []model.PermAccount, username string) []model.PermAccount {
-	matched := make([]model.PermAccount, 0, len(accounts))
-	for i := range accounts {
-		account := accounts[i]
-		if account.Username == username {
-			matched = append(matched, account)
-		}
-	}
-	return matched
-}
diff --git a/pkg/handler/server_ssh.go b/pkg/handler/server_ssh.go
index 0f99c287..2d4b47ad 100644
--- a/pkg/handler/server_ssh.go
+++ b/pkg/handler/server_ssh.go
@@ -64,26 +64,35 @@ func (s *Server) SFTPHandler(sess ssh.Session) {
 	}
 	addr, _, _ := net.SplitHostPort(sess.RemoteAddr().String())
 	directReq := sess.Context().Value(auth.ContextKeyDirectLoginFormat)
+	preparedDirectSFTP, _ := sess.Context().Value(auth.ContextKeyPreparedDirectSFTP).(*srvconn.PreparedDirectSFTP)
 	var sftpHandler *SftpHandler
 	termConf := s.GetTerminalConfig()
 	if directRequest, ok2 := directReq.(*auth.DirectLoginAssetReq); ok2 {
-		selectedAssets, err := s.getMatchedAssetsByDirectReq(currentUser, directRequest)
-		if err != nil {
-			logger.Errorf("Get matched assets failed: %s", err)
-			return
-		}
-		if directRequest.IsToken() && config.GetConf().ConnectionTokenReusable {
-			tokenInfo := directRequest.ConnectToken
-			key := cache.CreateAddrCacheKey(sess.RemoteAddr(), tokenInfo.Id)
-			// 缓存 token 信息
-			cache.TokenCacheInstance.Save(key, tokenInfo)
-			defer cache.TokenCacheInstance.Recycle(key)
-			logger.Infof("SFTP token key %s cached", key)
-		}
 		opts := buildDirectRequestOptions(currentUser, directRequest)
 		opts = append(opts, DirectConnectSftpMode(true))
-		opts = append(opts, DirectAssets(selectedAssets))
 		opts = append(opts, DirectTerminalConf(&termConf))
+		switch {
+		case preparedDirectSFTP != nil && preparedDirectSFTP.IsValid():
+			opts = append(opts, DirectFormatType(FormatToken))
+			opts = append(opts, DirectConnectToken(preparedDirectSFTP.Token))
+			opts = append(opts, DirectPreparedSFTP(preparedDirectSFTP))
+		case directRequest.IsToken():
+			if config.GetConf().ConnectionTokenReusable {
+				tokenInfo := directRequest.ConnectToken
+				key := cache.CreateAddrCacheKey(sess.RemoteAddr(), tokenInfo.Id)
+				// 缓存 token 信息
+				cache.TokenCacheInstance.Save(key, tokenInfo)
+				defer cache.TokenCacheInstance.Recycle(key)
+				logger.Infof("SFTP token key %s cached", key)
+			}
+		default:
+			selectedAssets, err := s.getMatchedAssetsByDirectReq(currentUser, directRequest)
+			if err != nil {
+				logger.Errorf("Get matched assets failed: %s", err)
+				return
+			}
+			opts = append(opts, DirectAssets(selectedAssets))
+		}
 		directSrv := NewDirectHandler(sess, s.jmsService, opts...)
 		sftpHandler = directSrv.NewSFTPHandler()
 	} else {
@@ -701,33 +710,12 @@ func buildSSHClientOptions(asset *model.Asset, account *model.Account,
 }
 
 func (s *Server) getMatchedAssetsByDirectReq(user *model.User, req *auth.DirectLoginAssetReq) ([]model.PermAsset, error) {
-	var getUserPermAssets func() ([]model.PermAsset, error)
-	if common.IsUUID(req.AssetTarget) {
-		getUserPermAssets = func() ([]model.PermAsset, error) {
-			return s.jmsService.GetUserPermAssetsById(user.ID, req.AssetTarget)
-		}
-	} else {
-		getUserPermAssets = func() ([]model.PermAsset, error) {
-			return s.jmsService.GetUserPermAssetsByIP(user.ID, req.AssetTarget)
-		}
-	}
-	i18nLang := i18n.NewLang(user.Language)
-	assets, err := getUserPermAssets()
-	if err != nil {
-		logger.Errorf("Get user %s perm asset failed: %s", user.String(), err)
-		return nil, fmt.Errorf("match asset failed: %s", i18nLang.T("Core API failed"))
-	}
-	if len(assets) == 0 {
-		logger.Infof("User %s no perm for asset %s", user.String(), req.AssetTarget)
-		return nil, fmt.Errorf("match asset failed: %s", i18nLang.T("No found asset"))
-	}
-	return assets, nil
+	return auth.GetMatchedAssetsByDirectReq(s.jmsService, user, req)
 }
 
 func (s *Server) getMatchedAccounts(user *model.User, req *auth.DirectLoginAssetReq,
 	permAssetDetail model.PermAssetDetail) ([]model.PermAccount, error) {
-	matched := GetMatchedAccounts(permAssetDetail.PermedAccounts, req.AccountUsername)
-	return matched, nil
+	return auth.MatchAccountsByUsername(permAssetDetail.PermedAccounts, req.AccountUsername), nil
 }
 
 func buildDirectRequestOptions(user *model.User, directRequest *auth.DirectLoginAssetReq) []DirectOpt {
@@ -743,65 +731,7 @@ func buildDirectRequestOptions(user *model.User, directRequest *auth.DirectLogin
 }
 
 func (s *Server) buildConnectToken(ctx ssh.Context, user *model.User, req *auth.DirectLoginAssetReq) (*model.ConnectToken, error) {
-	selectedAssets, err := s.getMatchedAssetsByDirectReq(user, req)
-	if err != nil {
-		return nil, err
-	}
-	i18nLang := i18n.NewLang(user.Language)
-	if len(selectedAssets) != 1 {
-		msg := fmt.Sprintf(i18nLang.T("Must be unique asset for %s"), req.AssetTarget)
-		return nil, errors.New(msg)
-	}
-	permAssetDetail, err := s.jmsService.GetUserPermAssetDetailById(user.ID, selectedAssets[0].ID)
-	if err != nil {
-		msg := fmt.Sprintf(i18nLang.T("Must be unique asset for %s"), req.AssetTarget)
-		logger.Errorf("Get permAssetDetail failed: %s", err)
-		return nil, errors.New(msg)
-	}
-
-	matchedProtocol := req.Protocol == model.ProtocolSSH
-	assetSupportedSSH := permAssetDetail.SupportProtocol(model.ProtocolSSH)
-	if !matchedProtocol || !assetSupportedSSH {
-		msg := "not ssh asset connection"
-		logger.Errorf("Direct Request ssh failed: %s", msg)
-		return nil, errors.New(msg)
-	}
-
-	selectAccounts, err := s.getMatchedAccounts(user, req, permAssetDetail)
-	if err != nil {
-		return nil, err
-	}
-	if len(selectAccounts) != 1 {
-		msg := fmt.Sprintf(i18nLang.T("Must be unique account for %s"), req.AccountUsername)
-		logger.Error(msg)
-		return nil, errors.New(msg)
-	}
-	selectAccount := selectAccounts[0]
-	remoteAddr, _, _ := net.SplitHostPort(ctx.RemoteAddr().String())
-	sessReq := &service.SuperConnectTokenReq{
-		UserId:        user.ID,
-		AssetId:       permAssetDetail.ID,
-		Account:       selectAccount.Alias,
-		Protocol:      model.ProtocolSSH,
-		ConnectMethod: model.ProtocolSSH,
-		RemoteAddr:    remoteAddr,
-	}
-	// ssh 非交互式的直连格式，不支持资产的登录复核
-	tokenInfo, err := s.jmsService.CreateSuperConnectToken(sessReq)
-	if err != nil {
-		msg := err.Error()
-		if tokenInfo.Detail != "" {
-			msg = tokenInfo.Detail
-		}
-		logger.Errorf("Create super connect token failed: %s", msg)
-		return nil, err
-	}
-	connectToken, err := s.jmsService.GetConnectTokenInfo(tokenInfo.ID, true)
-	if err != nil {
-		logger.Errorf("Create super connect token err: %s", err)
-		return nil, err
-	}
-	return &connectToken, nil
+	return auth.BuildDirectConnectToken(ctx, s.jmsService, user, req)
 }
 
 func (s *Server) buildSSHClient(tokenInfo *model.ConnectToken) (*srvconn.SSHClient, error) {
diff --git a/pkg/srvconn/prepared_sftp.go b/pkg/srvconn/prepared_sftp.go
new file mode 100644
index 00000000..0ed1cf31
--- /dev/null
+++ b/pkg/srvconn/prepared_sftp.go
@@ -0,0 +1,25 @@
+package srvconn
+
+import "github.com/jumpserver-dev/sdk-go/model"
+
+type PreparedDirectSFTP struct {
+	Token              *model.ConnectToken
+	Client             *SSHClient
+	DisableIdleRecycle bool
+}
+
+func (p *PreparedDirectSFTP) IsValid() bool {
+	return p != nil && p.Token != nil && p.Client != nil
+}
+
+func WithPreparedDirectSFTP(prepared *PreparedDirectSFTP) UserSftpOption {
+	return func(o *userSftpOption) {
+		o.preparedDirectSFTP = prepared
+	}
+}
+
+func WithPreparedDirectSFTPAsset(prepared *PreparedDirectSFTP) FolderBuilderOption {
+	return func(info *folderOptions) {
+		info.preparedDirectSFTP = prepared
+	}
+}
diff --git a/pkg/srvconn/sftp_asset.go b/pkg/srvconn/sftp_asset.go
index ad6949ba..c5692726 100644
--- a/pkg/srvconn/sftp_asset.go
+++ b/pkg/srvconn/sftp_asset.go
@@ -681,10 +681,18 @@ func (ad *AssetDir) getNewSftpConn(connectToken *model.ConnectToken,
 		return nil, errNoSelectAsset
 	}
 	timeout := config.GlobalConfig.SSHTimeout
-	sshClient, err := NewSSHClientWithToken(connectToken, timeout)
-	if err != nil {
-		logger.Errorf("Get new SSH client err: %s", err)
-		return nil, err
+	sshClient := (*SSHClient)(nil)
+	disableIdleRecycle := false
+	if prepared := ad.opts.preparedDirectSFTP; prepared != nil && prepared.IsValid() {
+		sshClient = prepared.Client
+		disableIdleRecycle = prepared.DisableIdleRecycle
+		logger.Infof("Reuse prepared direct sftp client %s", sshClient)
+	} else {
+		sshClient, err = NewSSHClientWithToken(connectToken, timeout)
+		if err != nil {
+			logger.Errorf("Get new SSH client err: %s", err)
+			return nil, err
+		}
 	}
 	sess, err := sshClient.AcquireSession()
 	if err != nil {
@@ -729,25 +737,26 @@ func (ad *AssetDir) getNewSftpConn(connectToken *model.ConnectToken,
 	}
 	maxIdleInt := ad.opts.terminalCfg.MaxIdleTime
 	conn = &SftpConn{
-		sshClient:   sshClient,
-		sshSession:  sess,
-		permAccount: su,
-		rootDirPath: sftpRoot,
-		client:      sftpClient,
-		HomeDirPath: homeDirPath,
-		token:       connectToken,
-		maxIdleTime: time.Duration(maxIdleInt) * time.Minute,
+		sshClient:          sshClient,
+		sshSession:         sess,
+		permAccount:        su,
+		rootDirPath:        sftpRoot,
+		client:             sftpClient,
+		HomeDirPath:        homeDirPath,
+		token:              connectToken,
+		maxIdleTime:        time.Duration(maxIdleInt) * time.Minute,
+		disableIdleRecycle: disableIdleRecycle,
 	}
 	return conn, nil
 }
 
-func NewSSHClientWithToken(connectToken *model.ConnectToken, timeout int) (*SSHClient, error) {
+func NewSSHClientWithToken(connectToken *model.ConnectToken, timeout int, extraOpts ...SSHClientOption) (*SSHClient, error) {
 	asset := connectToken.Asset
 	account := connectToken.Account
 	username := account.Username
 	protocol := connectToken.Protocol
 
-	sshAuthOpts := make([]SSHClientOption, 0, 6)
+	sshAuthOpts := make([]SSHClientOption, 0, 6+len(extraOpts))
 	sshAuthOpts = append(sshAuthOpts, SSHClientUsername(username))
 	sshAuthOpts = append(sshAuthOpts, SSHClientHost(asset.Address))
 	sshAuthOpts = append(sshAuthOpts, SSHClientPort(asset.ProtocolPort(protocol)))
@@ -781,6 +790,7 @@ func NewSSHClientWithToken(connectToken *model.ConnectToken, timeout int) (*SSHC
 		proxyArgs = append(proxyArgs, proxyArg)
 		sshAuthOpts = append(sshAuthOpts, SSHClientProxyClient(proxyArgs...))
 	}
+	sshAuthOpts = append(sshAuthOpts, extraOpts...)
 	return NewSSHClient(sshAuthOpts...)
 }
 
diff --git a/pkg/srvconn/sftpconn.go b/pkg/srvconn/sftpconn.go
index fdebccb1..877c0ea1 100644
--- a/pkg/srvconn/sftpconn.go
+++ b/pkg/srvconn/sftpconn.go
@@ -395,6 +395,9 @@ func (u *UserSftpConn) generateSubFoldersFromToken(token *model.ConnectToken) ma
 	opts = append(opts, WithToken(token))
 	opts = append(opts, WithFromType(u.loginFrom))
 	opts = append(opts, WithTerminalConfig(u.opts.terminalCfg))
+	if u.opts.preparedDirectSFTP != nil && u.opts.preparedDirectSFTP.IsValid() {
+		opts = append(opts, WithPreparedDirectSFTPAsset(u.opts.preparedDirectSFTP))
+	}
 	assetDir := NewAssetDir(u.jmsService, u.User, opts...)
 	assetDir.isFromWebTerminal = true
 	assetDir.loadSystemUsers()
@@ -459,7 +462,8 @@ type userSftpOption struct {
 
 	accountUsername string
 
-	terminalCfg *model.TerminalConfig
+	terminalCfg        *model.TerminalConfig
+	preparedDirectSFTP *PreparedDirectSFTP
 }
 
 type UserSftpOption func(*userSftpOption)
diff --git a/pkg/srvconn/sftpfile.go b/pkg/srvconn/sftpfile.go
index ac4be7d6..1e74c5dc 100644
--- a/pkg/srvconn/sftpfile.go
+++ b/pkg/srvconn/sftpfile.go
@@ -106,7 +106,8 @@ type folderOptions struct {
 
 	accountUsername string
 
-	terminalCfg *model.TerminalConfig
+	terminalCfg        *model.TerminalConfig
+	preparedDirectSFTP *PreparedDirectSFTP
 }
 
 func WithFolderUsername(username string) FolderBuilderOption {
@@ -220,10 +221,11 @@ type SftpConn struct {
 	isClosed    bool
 	rootDirPath string
 
-	nextExpiredTime time.Time
-	refs            atomic.Int32
-	lock            sync.Mutex
-	maxIdleTime     time.Duration
+	nextExpiredTime    time.Time
+	refs               atomic.Int32
+	lock               sync.Mutex
+	maxIdleTime        time.Duration
+	disableIdleRecycle bool
 }
 
 func (s *SftpConn) IsExpired() bool {
@@ -234,6 +236,9 @@ func (s *SftpConn) IsExpired() bool {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	now := time.Now()
+	if s.disableIdleRecycle {
+		return s.token.ExpireAt.IsExpired(now)
+	}
 	return now.Sub(s.nextExpiredTime) > 0 || s.token.ExpireAt.IsExpired(now)
 }
 
diff --git a/pkg/srvconn/ssh.go b/pkg/srvconn/ssh.go
index a6740f06..7e600785 100644
--- a/pkg/srvconn/ssh.go
+++ b/pkg/srvconn/ssh.go
@@ -56,12 +56,12 @@ func (cfg *SSHClientOptions) AuthMethods() []gossh.AuthMethod {
 	if cfg.PrivateAuth != nil {
 		authMethods = append(authMethods, gossh.PublicKeys(cfg.PrivateAuth))
 	}
-	if cfg.Password != "" {
-		authMethods = append(authMethods, gossh.Password(cfg.Password))
-	}
 	if cfg.keyboardAuth != nil {
 		authMethods = append(authMethods, gossh.KeyboardInteractive(cfg.keyboardAuth))
 	}
+	if cfg.Password != "" {
+		authMethods = append(authMethods, gossh.Password(cfg.Password))
+	}
 	if cfg.keyboardAuth == nil && cfg.Password != "" {
 		cfg.keyboardAuth = func(user, instruction string, questions []string, echos []bool) (answers []string, err error) {
 			if len(questions) == 0 {