diff --git a/app/core/application.py b/app/core/application.py
index 16b074f..eede744 100644
--- a/app/core/application.py
+++ b/app/core/application.py
@@ -9,7 +9,7 @@ from app.config.config import settings, sync_initial_settings
 from app.database.connection import connect_to_db, disconnect_from_db
 from app.database.initialization import initialize_database
 from app.exception.exceptions import setup_exception_handlers
-from app.log.logger import get_application_logger
+from app.log.logger import get_application_logger, setup_access_logging
 from app.middleware.middleware import setup_middlewares
 from app.router.routes import setup_routers
 from app.scheduler.scheduled_tasks import start_scheduler, stop_scheduler
@@ -150,4 +150,7 @@ def create_app() -> FastAPI:
     # 配置路由
     setup_routers(app)
 
+    # 配置访问日志API密钥隐藏
+    setup_access_logging()
+
     return app
diff --git a/app/database/services.py b/app/database/services.py
index 0cd73c5..d6f1211 100644
--- a/app/database/services.py
+++ b/app/database/services.py
@@ -8,6 +8,7 @@ import json
 from app.database.connection import database
 from app.database.models import Settings, ErrorLog, RequestLog, FileRecord, FileState
 from app.log.logger import get_database_logger
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_database_logger()
 
@@ -143,7 +144,7 @@ async def add_error_log(
             )
         )
         await database.execute(query)
-        logger.info(f"Added error log for key: {gemini_key}")
+        logger.info(f"Added error log for key: {redact_key_for_logging(gemini_key)}")
         return True
     except Exception as e:
         logger.error(f"Failed to add error log: {str(e)}")
diff --git a/app/handler/retry_handler.py b/app/handler/retry_handler.py
index 0161e27..749e463 100644
--- a/app/handler/retry_handler.py
+++ b/app/handler/retry_handler.py
@@ -4,6 +4,7 @@ from typing import Callable, TypeVar
 
 from app.config.config import settings
 from app.log.logger import get_retry_logger
+from app.utils.helpers import redact_key_for_logging
 
 T = TypeVar("T")
 logger = get_retry_logger()
@@ -37,7 +38,7 @@ class RetryHandler:
                         new_key = await key_manager.handle_api_failure(old_key, retries)
                         if new_key:
                             kwargs[self.key_arg] = new_key
-                            logger.info(f"Switched to new API key: {new_key}")
+                            logger.info(f"Switched to new API key: {redact_key_for_logging(new_key)}")
                         else:
                             logger.error(f"No valid API key available after {retries} retries.")
                             break
diff --git a/app/log/logger.py b/app/log/logger.py
index b6472a5..5ea7600 100644
--- a/app/log/logger.py
+++ b/app/log/logger.py
@@ -1,6 +1,7 @@
 import logging
 import platform
 import sys
+import re
 from typing import Dict, Optional
 
 # ANSI转义序列颜色代码
@@ -12,6 +13,9 @@ COLORS = {
     "CRITICAL": "\033[1;31m",  # 红色加粗
 }
 
+
+from app.utils.helpers import redact_key_for_logging as _redact_key_for_logging
+
 # Windows系统启用ANSI支持
 if platform.system() == "Windows":
     import ctypes
@@ -35,6 +39,50 @@ class ColoredFormatter(logging.Formatter):
         return super().format(record)
 
 
+class AccessLogFormatter(logging.Formatter):
+    """
+    Custom access log formatter that redacts API keys in URLs
+    """
+
+    # API key patterns to match in URLs
+    API_KEY_PATTERNS = [
+        r'\bAIza[0-9A-Za-z_-]{35}',  # Google API keys (like Gemini)
+        r'\bsk-[0-9A-Za-z_-]{20,}',  # OpenAI and general sk- prefixed keys
+    ]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # Compile regex patterns for better performance
+        self.compiled_patterns = [re.compile(pattern) for pattern in self.API_KEY_PATTERNS]
+
+    def format(self, record):
+        # Format the record normally first
+        formatted_msg = super().format(record)
+
+        # Redact API keys in the formatted message
+        return self._redact_api_keys_in_message(formatted_msg)
+
+    def _redact_api_keys_in_message(self, message: str) -> str:
+        """
+        Replace API keys in log message with redacted versions
+        """
+        try:
+            for pattern in self.compiled_patterns:
+                def replace_key(match):
+                    key = match.group(0)
+                    return _redact_key_for_logging(key)
+
+                message = pattern.sub(replace_key, message)
+
+            return message
+        except Exception as e:
+            # Log the error but don't expose the original message in case it contains keys
+            import logging
+            logger = logging.getLogger(__name__)
+            logger.error(f"Error redacting API keys in access log: {e}")
+            return "[LOG_REDACTION_ERROR]"
+
+
 # 日志格式 - 使用 fileloc 并设置固定宽度 (例如 30)
 FORMATTER = ColoredFormatter(
     "%(asctime)s | %(levelname)-17s | %(fileloc)-30s | %(message)s"
@@ -235,3 +283,43 @@ def get_files_logger():
 def get_vertex_express_logger():
     return Logger.setup_logger("vertex_express")
 
+
+def setup_access_logging():
+    """
+    Configure uvicorn access logging with API key redaction
+
+    This function sets up a custom access log formatter that automatically
+    redacts API keys in HTTP access logs. It works by:
+
+    1. Intercepting uvicorn's access log messages
+    2. Using regex patterns to find API keys in URLs
+    3. Replacing them with redacted versions (first6...last6)
+
+    Supported API key formats:
+    - Google/Gemini API keys: AIza[35 chars]
+    - OpenAI API keys: sk-[48 chars]
+    - General sk- prefixed keys: sk-[20+ chars]
+
+    Usage:
+    - Automatically called in main.py when running with uvicorn
+    - For production deployment with gunicorn, ensure this is called in startup
+    """
+    # Get the uvicorn access logger
+    access_logger = logging.getLogger("uvicorn.access")
+
+    # Remove existing handlers to avoid duplicate logs
+    for handler in access_logger.handlers[:]:
+        access_logger.removeHandler(handler)
+
+    # Create new handler with our custom formatter that includes timestamp and log level
+    handler = logging.StreamHandler(sys.stdout)
+    access_formatter = AccessLogFormatter("%(asctime)s | %(levelname)-8s | %(message)s")
+    handler.setFormatter(access_formatter)
+
+    # Add the handler to uvicorn access logger
+    access_logger.addHandler(handler)
+    access_logger.setLevel(logging.INFO)
+    access_logger.propagate = False
+
+    return access_logger
+
diff --git a/app/router/config_routes.py b/app/router/config_routes.py
index 6c453ff..ee4af76 100644
--- a/app/router/config_routes.py
+++ b/app/router/config_routes.py
@@ -11,6 +11,7 @@ from pydantic import BaseModel, Field
 from app.core.security import verify_auth_token
 from app.log.logger import Logger, get_config_routes_logger
 from app.service.config.config_service import ConfigService
+from app.utils.helpers import redact_key_for_logging
 
 router = APIRouter(prefix="/api/config", tags=["config"])
 
@@ -63,10 +64,10 @@ class DeleteKeysRequest(BaseModel):
 async def delete_single_key(key_to_delete: str, request: Request):
     auth_token = request.cookies.get("auth_token")
     if not auth_token or not verify_auth_token(auth_token):
-        logger.warning(f"Unauthorized attempt to delete key: {key_to_delete}")
+        logger.warning(f"Unauthorized attempt to delete key: {redact_key_for_logging(key_to_delete)}")
         return RedirectResponse(url="/", status_code=302)
     try:
-        logger.info(f"Attempting to delete key: {key_to_delete}")
+        logger.info(f"Attempting to delete key: {redact_key_for_logging(key_to_delete)}")
         result = await ConfigService.delete_key(key_to_delete)
         if not result.get("success"):
             raise HTTPException(
@@ -79,7 +80,7 @@ async def delete_single_key(key_to_delete: str, request: Request):
     except HTTPException as e:
         raise e
     except Exception as e:
-        logger.error(f"Error deleting key '{key_to_delete}': {e}", exc_info=True)
+        logger.error(f"Error deleting key '{redact_key_for_logging(key_to_delete)}': {e}", exc_info=True)
         raise HTTPException(status_code=500, detail=f"Error deleting key: {str(e)}")
 
 
diff --git a/app/router/files_routes.py b/app/router/files_routes.py
index a6628e0..826d8e0 100644
--- a/app/router/files_routes.py
+++ b/app/router/files_routes.py
@@ -15,6 +15,7 @@ from app.log.logger import get_files_logger
 from app.core.security import SecurityService
 from app.service.files.files_service import get_files_service
 from app.service.files.file_upload_handler import get_upload_handler
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_files_logger()
 
@@ -202,7 +203,7 @@ async def handle_upload(
 ):
     """处理文件上传请求"""
     try:
-        logger.info(f"Handling upload request: {request.method} {upload_path}, key={key}")
+        logger.info(f"Handling upload request: {request.method} {upload_path}, key={redact_key_for_logging(key)}")
         
         # 從查詢參數獲取 upload_id
         upload_id = request.query_params.get("upload_id")
@@ -222,7 +223,7 @@ async def handle_upload(
         # 使用真實的 API key 構建完整的 Google 上傳 URL
         # 保留原始 URL 的所有參數，但使用真實的 API key
         upload_url = original_upload_url
-        logger.info(f"Using real API key for upload: {real_api_key[:8]}...{real_api_key[-4:]}")
+        logger.info(f"Using real API key for upload: {redact_key_for_logging(real_api_key)}")
         
         # 代理上传请求
         upload_handler = get_upload_handler()
diff --git a/app/router/gemini_routes.py b/app/router/gemini_routes.py
index 4910357..189c300 100644
--- a/app/router/gemini_routes.py
+++ b/app/router/gemini_routes.py
@@ -13,6 +13,7 @@ from app.service.model.model_service import ModelService
 from app.handler.retry_handler import RetryHandler
 from app.handler.error_handler import handle_route_errors
 from app.core.constants import API_VERSION
+from app.utils.helpers import redact_key_for_logging
 
 router = APIRouter(prefix=f"/gemini/{API_VERSION}")
 router_v1beta = APIRouter(prefix=f"/{API_VERSION}")
@@ -52,7 +53,7 @@ async def list_models(
         api_key = await key_manager.get_first_valid_key()
         if not api_key:
             raise HTTPException(status_code=503, detail="No valid API keys available to fetch models.")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         models_data = await model_service.get_gemini_models(api_key)
         if not models_data or "models" not in models_data:
@@ -125,7 +126,7 @@ async def generate_content(
                 logger.info(f"TTS responseModalities: {response_modalities}")
                 logger.info(f"TTS speechConfig: {speech_config}")
 
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         if not await model_service.check_model_support(model_name):
             raise HTTPException(status_code=400, detail=f"Model {model_name} is not supported")
@@ -169,7 +170,7 @@ async def stream_generate_content(
     async with handle_route_errors(logger, operation_name, failure_message="Streaming request initiation failed"):
         logger.info(f"Handling Gemini streaming content generation for model: {model_name}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         if not await model_service.check_model_support(model_name):
             raise HTTPException(status_code=400, detail=f"Model {model_name} is not supported")
@@ -198,7 +199,7 @@ async def count_tokens(
     async with handle_route_errors(logger, operation_name, failure_message="Token counting failed"):
         logger.info(f"Handling Gemini token count request for model: {model_name}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         if not await model_service.check_model_support(model_name):
             raise HTTPException(status_code=400, detail=f"Model {model_name} is not supported")
@@ -274,9 +275,9 @@ async def reset_selected_key_fail_counts(
                 if result:
                     reset_count += 1
                 else:
-                    logger.warning(f"Key not found during selective reset: {key}")
+                    logger.warning(f"Key not found during selective reset: {redact_key_for_logging(key)}")
             except Exception as key_error:
-                logger.error(f"Error resetting key {key}: {str(key_error)}")
+                logger.error(f"Error resetting key {redact_key_for_logging(key)}: {str(key_error)}")
                 errors.append(f"Key {key}: {str(key_error)}")
 
         if errors:
@@ -303,7 +304,7 @@ async def reset_selected_key_fail_counts(
 async def reset_key_fail_count(api_key: str, key_manager: KeyManager = Depends(get_key_manager)):
     """重置指定Gemini API密钥的失败计数"""
     logger.info("-" * 50 + "reset_gemini_key_fail_count" + "-" * 50)
-    logger.info(f"Resetting failure count for API key: {api_key}")
+    logger.info(f"Resetting failure count for API key: {redact_key_for_logging(api_key)}")
     
     try:
         result = await key_manager.reset_key_failure_count(api_key)
@@ -348,7 +349,7 @@ async def verify_key(api_key: str, chat_service: GeminiChatService = Depends(get
         async with key_manager.failure_count_lock:
             if api_key in key_manager.key_failure_counts:
                 key_manager.key_failure_counts[api_key] += 1
-                logger.warning(f"Verification exception for key: {api_key}, incrementing failure count")
+                logger.warning(f"Verification exception for key: {redact_key_for_logging(api_key)}, incrementing failure count")
         
         return JSONResponse({"status": "invalid", "error": str(e)})
 
@@ -389,14 +390,14 @@ async def verify_selected_keys(
             return api_key, "valid", None
         except Exception as e:
             error_message = str(e)
-            logger.warning(f"Key verification failed for {api_key}: {error_message}")
+            logger.warning(f"Key verification failed for {redact_key_for_logging(api_key)}: {error_message}")
             async with key_manager.failure_count_lock:
                 if api_key in key_manager.key_failure_counts:
                     key_manager.key_failure_counts[api_key] += 1
-                    logger.warning(f"Bulk verification exception for key: {api_key}, incrementing failure count")
+                    logger.warning(f"Bulk verification exception for key: {redact_key_for_logging(api_key)}, incrementing failure count")
                 else:
                      key_manager.key_failure_counts[api_key] = 1
-                     logger.warning(f"Bulk verification exception for key: {api_key}, initializing failure count to 1")
+                     logger.warning(f"Bulk verification exception for key: {redact_key_for_logging(api_key)}, initializing failure count to 1")
             failed_keys[api_key] = error_message
             return api_key, "invalid", error_message
 
diff --git a/app/router/openai_compatiable_routes.py b/app/router/openai_compatiable_routes.py
index 68c4cbe..92730d0 100644
--- a/app/router/openai_compatiable_routes.py
+++ b/app/router/openai_compatiable_routes.py
@@ -13,6 +13,7 @@ from app.handler.error_handler import handle_route_errors
 from app.log.logger import get_openai_compatible_logger
 from app.service.key.key_manager import KeyManager, get_key_manager_instance
 from app.service.openai_compatiable.openai_compatiable_service import OpenAICompatiableService
+from app.utils.helpers import redact_key_for_logging
 
 
 router = APIRouter()
@@ -46,7 +47,7 @@ async def list_models(
     async with handle_route_errors(logger, operation_name):
         logger.info("Handling models list request")
         api_key = await key_manager.get_first_valid_key()
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
         return await openai_service.get_models(api_key)
 
 
@@ -69,7 +70,7 @@ async def chat_completion(
     async with handle_route_errors(logger, operation_name):
         logger.info(f"Handling chat completion request for model: {request.model}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {current_api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(current_api_key)}")
 
         if is_image_chat:
             response = await openai_service.create_image_chat_completion(request, current_api_key)
@@ -107,7 +108,7 @@ async def embedding(
     async with handle_route_errors(logger, operation_name):
         logger.info(f"Handling embedding request for model: {request.model}")
         api_key = await key_manager.get_next_working_key()
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
         return await openai_service.create_embeddings(
             input_text=request.input, model=request.model, api_key=api_key
         )
diff --git a/app/router/openai_routes.py b/app/router/openai_routes.py
index 6bee7c2..e6cf9ae 100644
--- a/app/router/openai_routes.py
+++ b/app/router/openai_routes.py
@@ -18,6 +18,7 @@ from app.service.image.image_create_service import ImageCreateService
 from app.service.tts.tts_service import TTSService
 from app.service.key.key_manager import KeyManager, get_key_manager_instance
 from app.service.model.model_service import ModelService
+from app.utils.helpers import redact_key_for_logging
 
 router = APIRouter()
 logger = get_openai_logger()
@@ -60,7 +61,7 @@ async def list_models(
     async with handle_route_errors(logger, operation_name):
         logger.info("Handling models list request")
         api_key = await key_manager.get_first_valid_key()
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
         return await model_service.get_gemini_openai_models(api_key)
 
 
@@ -84,7 +85,7 @@ async def chat_completion(
     async with handle_route_errors(logger, operation_name):
         logger.info(f"Handling chat completion request for model: {request.model}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {current_api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(current_api_key)}")
 
         if not await model_service.check_model_support(request.model):
             raise HTTPException(
@@ -129,7 +130,7 @@ async def embedding(
     async with handle_route_errors(logger, operation_name):
         logger.info(f"Handling embedding request for model: {request.model}")
         api_key = await key_manager.get_next_working_key()
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
         response = await embedding_service.create_embedding(
             input_text=request.input, model=request.model, api_key=api_key
         )
@@ -170,6 +171,6 @@ async def text_to_speech(
     async with handle_route_errors(logger, operation_name):
         logger.info(f"Handling TTS request for model: {request.model}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
         audio_data = await tts_service.create_tts(request, api_key)
         return Response(content=audio_data, media_type="audio/wav")
diff --git a/app/router/stats_routes.py b/app/router/stats_routes.py
index 32658e9..5a3d74e 100644
--- a/app/router/stats_routes.py
+++ b/app/router/stats_routes.py
@@ -3,6 +3,7 @@ from starlette import status
 from app.core.security import verify_auth_token
 from app.service.stats.stats_service import StatsService
 from app.log.logger import get_stats_logger
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_stats_logger()
 
@@ -48,7 +49,7 @@ async def get_key_usage_details(key: str):
             return {}
         return usage_details
     except Exception as e:
-        logger.error(f"Error fetching key usage details for key {key[:4]}...: {e}")
+        logger.error(f"Error fetching key usage details for key {redact_key_for_logging(key)}: {e}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"获取密钥使用详情时出错: {e}"
diff --git a/app/router/vertex_express_routes.py b/app/router/vertex_express_routes.py
index 3b7f0f8..b4d57c8 100644
--- a/app/router/vertex_express_routes.py
+++ b/app/router/vertex_express_routes.py
@@ -11,6 +11,7 @@ from app.service.model.model_service import ModelService
 from app.handler.retry_handler import RetryHandler
 from app.handler.error_handler import handle_route_errors
 from app.core.constants import API_VERSION
+from app.utils.helpers import redact_key_for_logging
 
 router = APIRouter(prefix=f"/vertex-express/{API_VERSION}")
 logger = get_vertex_express_logger()
@@ -48,7 +49,7 @@ async def list_models(
         api_key = await key_manager.get_first_valid_key()
         if not api_key:
             raise HTTPException(status_code=503, detail="No valid API keys available to fetch models.")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         models_data = await model_service.get_gemini_models(api_key)
         if not models_data or "models" not in models_data:
@@ -105,7 +106,7 @@ async def generate_content(
     async with handle_route_errors(logger, operation_name, failure_message="Content generation failed"):
         logger.info(f"Handling Gemini content generation request for model: {model_name}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         if not await model_service.check_model_support(model_name):
             raise HTTPException(status_code=400, detail=f"Model {model_name} is not supported")
@@ -133,7 +134,7 @@ async def stream_generate_content(
     async with handle_route_errors(logger, operation_name, failure_message="Streaming request initiation failed"):
         logger.info(f"Handling Gemini streaming content generation for model: {model_name}")
         logger.debug(f"Request: \n{request.model_dump_json(indent=2)}")
-        logger.info(f"Using API key: {api_key}")
+        logger.info(f"Using API key: {redact_key_for_logging(api_key)}")
 
         if not await model_service.check_model_support(model_name):
             raise HTTPException(status_code=400, detail=f"Model {model_name} is not supported")
diff --git a/app/scheduler/scheduled_tasks.py b/app/scheduler/scheduled_tasks.py
index b185f1e..de2413e 100644
--- a/app/scheduler/scheduled_tasks.py
+++ b/app/scheduler/scheduled_tasks.py
@@ -9,6 +9,7 @@ from app.service.error_log.error_log_service import delete_old_error_logs
 from app.service.key.key_manager import get_key_manager_instance
 from app.service.request_log.request_log_service import delete_old_request_logs_task
 from app.service.files.files_service import get_files_service
+from app.utils.helpers import redact_key_for_logging
 
 logger = Logger.setup_logger("scheduler")
 
@@ -51,7 +52,7 @@ async def check_failed_keys():
 
         for key in keys_to_check:
             # 隐藏部分 key 用于日志记录
-            log_key = f"{key[:4]}...{key[-4:]}" if len(key) > 8 else key
+            log_key = redact_key_for_logging(key)
             logger.info(f"Verifying key: {log_key}...")
             try:
                 # 构造测试请求
diff --git a/app/service/chat/gemini_chat_service.py b/app/service/chat/gemini_chat_service.py
index 1444146..0845b57 100644
--- a/app/service/chat/gemini_chat_service.py
+++ b/app/service/chat/gemini_chat_service.py
@@ -14,6 +14,7 @@ from app.log.logger import get_gemini_logger
 from app.service.client.api_client import GeminiApiClient
 from app.service.key.key_manager import KeyManager
 from app.database.services import add_error_log, add_request_log, get_file_api_key
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_gemini_logger()
 
@@ -296,10 +297,10 @@ class GeminiChatService:
             logger.info(f"Request contains file references: {file_names}")
             file_api_key = await get_file_api_key(file_names[0])
             if file_api_key:
-                logger.info(f"Found API key for file {file_names[0]}: {file_api_key[:8]}...{file_api_key[-4:]}")
+                logger.info(f"Found API key for file {file_names[0]}: {redact_key_for_logging(file_api_key)}")
                 api_key = file_api_key  # 使用文件的 API key
             else:
-                logger.warning(f"No API key found for file {file_names[0]}, using default key: {api_key[:8]}...{api_key[-4:]}")
+                logger.warning(f"No API key found for file {file_names[0]}, using default key: {redact_key_for_logging(api_key)}")
         
         payload = _build_payload(model, request)
         start_time = time.perf_counter()
@@ -402,10 +403,10 @@ class GeminiChatService:
             logger.info(f"Request contains file references: {file_names}")
             file_api_key = await get_file_api_key(file_names[0])
             if file_api_key:
-                logger.info(f"Found API key for file {file_names[0]}: {file_api_key[:8]}...{file_api_key[-4:]}")
+                logger.info(f"Found API key for file {file_names[0]}: {redact_key_for_logging(file_api_key)}")
                 api_key = file_api_key  # 使用文件的 API key
             else:
-                logger.warning(f"No API key found for file {file_names[0]}, using default key: {api_key[:8]}...{api_key[-4:]}")
+                logger.warning(f"No API key found for file {file_names[0]}, using default key: {redact_key_for_logging(api_key)}")
                 
         retries = 0
         max_retries = settings.MAX_RETRIES
@@ -472,7 +473,7 @@ class GeminiChatService:
 
                 api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries)
                 if api_key:
-                    logger.info(f"Switched to new API key: {api_key}")
+                    logger.info(f"Switched to new API key: {redact_key_for_logging(api_key)}")
                 else:
                     logger.error(f"No valid API key available after {retries} retries.")
                     break
diff --git a/app/service/chat/vertex_express_chat_service.py b/app/service/chat/vertex_express_chat_service.py
index 95f1b77..6885ac0 100644
--- a/app/service/chat/vertex_express_chat_service.py
+++ b/app/service/chat/vertex_express_chat_service.py
@@ -14,6 +14,7 @@ from app.log.logger import get_gemini_logger
 from app.service.client.api_client import GeminiApiClient
 from app.service.key.key_manager import KeyManager
 from app.database.services import add_error_log, add_request_log
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_gemini_logger()
 
@@ -340,7 +341,7 @@ class GeminiChatService:
 
                 api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries)
                 if api_key:
-                    logger.info(f"Switched to new API key: {api_key}")
+                    logger.info(f"Switched to new API key: {redact_key_for_logging(api_key)}")
                 else:
                     logger.error(f"No valid API key available after {retries} retries.")
                     break
diff --git a/app/service/files/file_upload_handler.py b/app/service/files/file_upload_handler.py
index c1b919c..7a33525 100644
--- a/app/service/files/file_upload_handler.py
+++ b/app/service/files/file_upload_handler.py
@@ -12,6 +12,7 @@ from app.config.config import settings
 from app.database import services as db_services
 from app.database.models import FileState
 from app.log.logger import get_files_logger
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_files_logger()
 
@@ -144,7 +145,7 @@ class FileUploadHandler:
                                     sha256_hash=file_data.get("sha256Hash"),
                                     user_token=session_info["user_token"]
                                 )
-                                logger.info(f"Created file record: name={real_file_name}, api_key={session_info['api_key'][:8]}...{session_info['api_key'][-4:]}")
+                                logger.info(f"Created file record: name={real_file_name}, api_key={redact_key_for_logging(session_info['api_key'])}")
                             else:
                                 logger.warning(f"No upload session found for URL: {upload_url}")
                         else:
diff --git a/app/service/files/files_service.py b/app/service/files/files_service.py
index c33ede4..332cd19 100644
--- a/app/service/files/files_service.py
+++ b/app/service/files/files_service.py
@@ -13,6 +13,7 @@ from app.database.models import FileState
 from app.domain.file_models import FileMetadata, ListFilesResponse
 from fastapi import HTTPException
 from app.log.logger import get_files_logger
+from app.utils.helpers import redact_key_for_logging
 from app.service.client.api_client import GeminiApiClient
 from app.service.key.key_manager import get_key_manager_instance
 
@@ -102,7 +103,7 @@ class FilesService:
                 
                 # 儲存上傳資訊到 headers 中，供後續使用
                 # 不在這裡創建數據庫記錄，等到上傳完成後再創建
-                logger.info(f"Upload initialized with API key: {api_key[:8]}...{api_key[-4:]}")
+                logger.info(f"Upload initialized with API key: {redact_key_for_logging(api_key)}")
                 
                 # 解析响应 - 初始化响应可能是空的
                 response_data = {}
@@ -133,7 +134,7 @@ class FilesService:
                             "created_at": datetime.now(timezone.utc),
                             "upload_url": upload_url
                         }
-                        logger.info(f"Stored upload session for upload_id={upload_id}: api_key={api_key[:8]}...{api_key[-4:]}")
+                        logger.info(f"Stored upload session for upload_id={upload_id}: api_key={redact_key_for_logging(api_key)}")
                         logger.debug(f"Total active sessions: {len(_upload_sessions)}")
                 else:
                     logger.warning(f"No upload_id found in upload URL: {upload_url}")
@@ -202,7 +203,7 @@ class FilesService:
             # 先嘗試直接查找
             session = _upload_sessions.get(key)
             if session:
-                logger.debug(f"Found session by direct key {key}")
+                logger.debug(f"Found session by direct key {redact_key_for_logging(key)}")
                 return session
             
             # 如果是 URL，嘗試提取 upload_id
@@ -217,7 +218,7 @@ class FilesService:
                         logger.debug(f"Found session by upload_id {upload_id} from URL")
                         return session
             
-            logger.debug(f"No session found for key: {key}")
+            logger.debug(f"No session found for key: {redact_key_for_logging(key)}")
             return None
     
     async def get_file(self, file_name: str, user_token: str) -> FileMetadata:
diff --git a/app/service/key/key_manager.py b/app/service/key/key_manager.py
index 73d897b..0925ba9 100644
--- a/app/service/key/key_manager.py
+++ b/app/service/key/key_manager.py
@@ -4,6 +4,7 @@ from typing import Dict, Union
 
 from app.config.config import settings
 from app.log.logger import get_key_manager_logger
+from app.utils.helpers import redact_key_for_logging
 
 logger = get_key_manager_logger()
 
@@ -65,7 +66,7 @@ class KeyManager:
         async with self.failure_count_lock:
             if key in self.key_failure_counts:
                 self.key_failure_counts[key] = 0
-                logger.info(f"Reset failure count for key: {key}")
+                logger.info(f"Reset failure count for key: {redact_key_for_logging(key)}")
                 return True
             logger.warning(
                 f"Attempt to reset failure count for non-existent key: {key}"
@@ -77,7 +78,7 @@ class KeyManager:
         async with self.vertex_failure_count_lock:
             if key in self.vertex_key_failure_counts:
                 self.vertex_key_failure_counts[key] = 0
-                logger.info(f"Reset failure count for Vertex key: {key}")
+                logger.info(f"Reset failure count for Vertex key: {redact_key_for_logging(key)}")
                 return True
             logger.warning(
                 f"Attempt to reset failure count for non-existent Vertex key: {key}"
@@ -116,7 +117,7 @@ class KeyManager:
             self.key_failure_counts[api_key] += 1
             if self.key_failure_counts[api_key] >= self.MAX_FAILURES:
                 logger.warning(
-                    f"API key {api_key} has failed {self.MAX_FAILURES} times"
+                    f"API key {redact_key_for_logging(api_key)} has failed {self.MAX_FAILURES} times"
                 )
         if retries < settings.MAX_RETRIES:
             return await self.get_next_working_key()
@@ -129,7 +130,7 @@ class KeyManager:
             self.vertex_key_failure_counts[api_key] += 1
             if self.vertex_key_failure_counts[api_key] >= self.MAX_FAILURES:
                 logger.warning(
-                    f"Vertex Express API key {api_key} has failed {self.MAX_FAILURES} times"
+                    f"Vertex Express API key {redact_key_for_logging(api_key)} has failed {self.MAX_FAILURES} times"
                 )
 
     def get_fail_count(self, key: str) -> int:
diff --git a/app/service/openai_compatiable/openai_compatiable_service.py b/app/service/openai_compatiable/openai_compatiable_service.py
index 51e062b..d50b7da 100644
--- a/app/service/openai_compatiable/openai_compatiable_service.py
+++ b/app/service/openai_compatiable/openai_compatiable_service.py
@@ -13,6 +13,7 @@ from app.database.services import (
 from app.domain.openai_models import ChatRequest, ImageGenerationRequest
 from app.service.client.api_client import OpenaiApiClient
 from app.service.key.key_manager import KeyManager
+from app.utils.helpers import redact_key_for_logging
 from app.log.logger import get_openai_compatible_logger
 
 logger = get_openai_compatible_logger()
@@ -159,7 +160,7 @@ class OpenAICompatiableService:
                         current_attempt_key, retries
                     )
                     if api_key:
-                        logger.info(f"Switched to new API key: {api_key}")
+                        logger.info(f"Switched to new API key: {redact_key_for_logging(api_key)}")
                     else:
                         logger.error(
                             f"No valid API key available after {retries} retries."
diff --git a/app/utils/helpers.py b/app/utils/helpers.py
index 7a3cc90..c283654 100644
--- a/app/utils/helpers.py
+++ b/app/utils/helpers.py
@@ -154,6 +154,25 @@ def is_valid_api_key(key: str) -> bool:
 
 
 
+def redact_key_for_logging(key: str) -> str:
+    """
+    Redacts API key for secure logging by showing only first and last 6 characters.
+
+    Args:
+        key: API key to redact
+
+    Returns:
+        str: Redacted key in format "first6...last6" or descriptive placeholder for edge cases
+    """
+    if not key:
+        return key
+
+    if len(key) <= 12:
+        return f"{key[:3]}...{key[-3:]}"
+    else:
+        return f"{key[:6]}...{key[-6:]}"
+
+
 def get_current_version(default_version: str = "0.0.0") -> str:
     """Reads the current version from the VERSION file."""
     version_file = VERSION_FILE_PATH
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..739954c
--- /dev/null
+++ b/tests/__init__.py
@@ -0,0 +1 @@
+# Tests package
\ No newline at end of file
diff --git a/tests/test_key_redaction.py b/tests/test_key_redaction.py
new file mode 100644
index 0000000..d1ef0e7
--- /dev/null
+++ b/tests/test_key_redaction.py
@@ -0,0 +1,187 @@
+"""
+Unit tests for API key redaction functionality
+"""
+
+import unittest
+import logging
+from unittest.mock import patch, MagicMock
+
+from app.utils.helpers import redact_key_for_logging
+from app.log.logger import AccessLogFormatter
+
+
+class TestKeyRedaction(unittest.TestCase):
+    """Test cases for the redact_key_for_logging function"""
+
+    def test_valid_long_key_redaction(self):
+        """Test redaction of valid long API keys"""
+        # Test Google/Gemini API key
+        # This value is a random generated string for testing
+        gemini_key = "AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI"
+        result = redact_key_for_logging(gemini_key)
+        expected = "AIzaSy...xDfGhI"
+        self.assertEqual(result, expected)
+
+        # Test OpenAI API key
+        # This value is a random generated string for testing
+        openai_key = "sk-1234567890abcdef1234567890abcdef1234567890abcdef"
+        result = redact_key_for_logging(openai_key)
+        expected = "sk-123...abcdef"
+        self.assertEqual(result, expected)
+
+    def test_short_key_handling(self):
+        """Test handling of short keys"""
+        short_key = "short"
+        result = redact_key_for_logging(short_key)
+        self.assertEqual(result, "[SHORT_KEY]")
+
+        # Test exactly 12 characters (boundary case)
+        boundary_key = "123456789012"
+        result = redact_key_for_logging(boundary_key)
+        self.assertEqual(result, "[SHORT_KEY]")
+
+    def test_empty_and_none_keys(self):
+        """Test handling of empty and None keys"""
+        # Test empty string
+        result = redact_key_for_logging("")
+        self.assertEqual(result, "[INVALID_KEY]")
+
+        # Test None
+        result = redact_key_for_logging(None)
+        self.assertEqual(result, "[INVALID_KEY]")
+
+    def test_invalid_input_types(self):
+        """Test handling of invalid input types"""
+        # Test integer
+        result = redact_key_for_logging(123)
+        self.assertEqual(result, "[INVALID_KEY]")
+
+        # Test list
+        result = redact_key_for_logging(["key"])
+        self.assertEqual(result, "[INVALID_KEY]")
+
+        # Test dict
+        result = redact_key_for_logging({"key": "value"})
+        self.assertEqual(result, "[INVALID_KEY]")
+
+    def test_boundary_cases(self):
+        """Test boundary cases for key length"""
+        # Test 13 characters (just above the threshold)
+        key_13 = "1234567890123"
+        result = redact_key_for_logging(key_13)
+        expected = "123456...890123"
+        self.assertEqual(result, expected)
+
+        # Test very long key
+        long_key = "a" * 100
+        result = redact_key_for_logging(long_key)
+        expected = "aaaaaa...aaaaaa"
+        self.assertEqual(result, expected)
+
+
+class TestAccessLogFormatter(unittest.TestCase):
+    """Test cases for the AccessLogFormatter class"""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        self.formatter = AccessLogFormatter()
+
+    def test_gemini_key_redaction_in_url(self):
+        """Test redaction of Gemini API keys in URLs"""
+        log_message = (
+            'POST /verify-key/AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI HTTP/1.1" 200'
+        )
+        result = self.formatter._redact_api_keys_in_message(log_message)
+        self.assertIn("AIzaSy...xDfGhI", result)
+        self.assertNotIn("AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI", result)
+
+    def test_openai_key_redaction_in_url(self):
+        """Test redaction of OpenAI API keys in URLs"""
+        log_message = 'GET /api/models?key=sk-1234567890abcdef1234567890abcdef1234567890abcdef HTTP/1.1" 200'
+        result = self.formatter._redact_api_keys_in_message(log_message)
+        self.assertIn("sk-123...abcdef", result)
+        self.assertNotIn("sk-1234567890abcdef1234567890abcdef1234567890abcdef", result)
+
+    def test_multiple_keys_in_message(self):
+        """Test redaction of multiple API keys in a single message"""
+        log_message = "Request with keys: AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI and sk-1234567890abcdef1234567890abcdef1234567890abcdef"
+        result = self.formatter._redact_api_keys_in_message(log_message)
+        self.assertIn("AIzaSy...xDfGhI", result)
+        self.assertIn("sk-123...abcdef", result)
+        self.assertNotIn("AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI", result)
+        self.assertNotIn("sk-1234567890abcdef1234567890abcdef1234567890abcdef", result)
+
+    def test_no_keys_in_message(self):
+        """Test that messages without API keys are unchanged"""
+        log_message = 'GET /api/health HTTP/1.1" 200'
+        result = self.formatter._redact_api_keys_in_message(log_message)
+        self.assertEqual(result, log_message)
+
+    def test_partial_key_patterns_not_redacted(self):
+        """Test that partial key patterns are not redacted"""
+        log_message = "Message with partial patterns: AIza sk- incomplete"
+        result = self.formatter._redact_api_keys_in_message(log_message)
+        self.assertEqual(result, log_message)
+
+    def test_error_handling_in_redaction(self):
+        """Test error handling in the redaction process"""
+        # Test by directly calling _redact_api_keys_in_message with a broken pattern
+        original_patterns = self.formatter.compiled_patterns
+        # Create a mock pattern that will raise an exception
+        mock_pattern = MagicMock()
+        mock_pattern.sub.side_effect = Exception("Regex error")
+        self.formatter.compiled_patterns = [mock_pattern]
+
+        try:
+            log_message = (
+                'POST /verify-key/AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI HTTP/1.1" 200'
+            )
+            result = self.formatter._redact_api_keys_in_message(log_message)
+            self.assertEqual(result, "[LOG_REDACTION_ERROR]")
+        finally:
+            # Restore original patterns
+            self.formatter.compiled_patterns = original_patterns
+
+    def test_format_method(self):
+        """Test the format method of AccessLogFormatter"""
+        # Create a mock log record
+        record = MagicMock()
+        record.getMessage.return_value = (
+            'POST /verify-key/AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI HTTP/1.1" 200'
+        )
+
+        # Mock the parent format method
+        with patch(
+            "logging.Formatter.format",
+            return_value='2025-01-01 12:00:00 | INFO | POST /verify-key/AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI HTTP/1.1" 200',
+        ):
+            result = self.formatter.format(record)
+            self.assertIn("AIzaSy...xDfGhI", result)
+            self.assertNotIn("AIzaSyDhKGfJ8xYzQwErTyUiOpLkMnBvCxDfGhI", result)
+
+    def test_regex_patterns_compilation(self):
+        """Test that regex patterns are properly compiled"""
+        formatter = AccessLogFormatter()
+        self.assertEqual(len(formatter.compiled_patterns), 2)
+        self.assertTrue(
+            all(hasattr(pattern, "sub") for pattern in formatter.compiled_patterns)
+        )
+
+    def test_flexible_openai_pattern(self):
+        """Test the flexible OpenAI pattern matches various formats"""
+        test_cases = [
+            "sk-1234567890abcdef1234567890abcdef1234567890abcdef",  # Standard 48 chars
+            "sk-proj-1234567890abcdef1234567890abcdef1234567890abcdef",  # Project key
+            "sk-1234567890abcdef_1234567890abcdef-1234567890abcdef",  # With underscores/hyphens
+            "sk-12345678901234567890",  # Shorter key (20 chars)
+        ]
+
+        for test_key in test_cases:
+            log_message = f"Request with key: {test_key}"
+            result = self.formatter._redact_api_keys_in_message(log_message)
+            self.assertNotIn(test_key, result)
+            self.assertIn("sk-", result)  # Should still contain the prefix
+
+
+if __name__ == "__main__":
+    unittest.main()