base apis

api/apps/__init__.py

@@ -13,13 +13,12 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
+import logging
 import os
 import sys
-import logging
 from importlib.util import module_from_spec, spec_from_file_location
 from pathlib import Path
 from quart import Blueprint, Quart, request, g, current_app, session
-from werkzeug.wrappers.request import Request
 from flasgger import Swagger
 from itsdangerous.url_safe import URLSafeTimedSerializer as Serializer
 from quart_cors import cors
@@ -40,7 +39,6 @@ settings.init_settings()
 
 __all__ = ["app"]
 
-Request.json = property(lambda self: self.get_json(force=True, silent=True))
 
 app = Quart(__name__)
 app = cors(app, allow_origin="*")

api/apps/api_app.py

@@ -18,8 +18,7 @@ from quart import request
 from api.db.db_models import APIToken
 from api.db.services.api_service import APITokenService, API4ConversationService
 from api.db.services.user_service import UserTenantService
-from api.utils.api_utils import server_error_response, get_data_error_result, get_json_result, validate_request, \
+from api.utils.api_utils import generate_confirmation_token, get_data_error_result, get_json_result, request_json, server_error_response, validate_request
-    generate_confirmation_token
 from common.time_utils import current_timestamp, datetime_format
 from api.apps import login_required, current_user
 
@@ -27,7 +26,7 @@ from api.apps import login_required, current_user
 @manager.route('/new_token', methods=['POST'])  # noqa: F821
 @login_required
 async def new_token():
-    req = await request.json
+    req = await request_json()
     try:
         tenants = UserTenantService.query(user_id=current_user.id)
         if not tenants:
@@ -73,7 +72,7 @@ def token_list():
 @validate_request("tokens", "tenant_id")
 @login_required
 async def rm():
-    req = await request.json
+    req = await request_json()
     try:
         for token in req["tokens"]:
             APITokenService.filter_delete(
@@ -116,4 +115,3 @@ def stats():
         return get_json_result(data=res)
     except Exception as e:
         return server_error_response(e)
-

api/apps/auth/github.py

@@ -14,7 +14,7 @@
 #  limitations under the License.
 #
 
-import requests
+from common.http_client import async_request, sync_request
 from .oauth import OAuthClient, UserInfo
 
 
@@ -34,24 +34,49 @@ class GithubOAuthClient(OAuthClient):
 
     def fetch_user_info(self, access_token, **kwargs):
         """
-        Fetch GitHub user info.
+        Fetch GitHub user info (synchronous).
         """
         user_info = {}
         try:
             headers = {"Authorization": f"Bearer {access_token}"}
-            # user info
+            response = sync_request("GET", self.userinfo_url, headers=headers, timeout=self.http_request_timeout)
-            response = requests.get(self.userinfo_url, headers=headers, timeout=self.http_request_timeout)
             response.raise_for_status()
             user_info.update(response.json())
-            # email info
+            email_response = sync_request(
-            response = requests.get(self.userinfo_url+"/emails", headers=headers, timeout=self.http_request_timeout)
+                "GET", self.userinfo_url + "/emails", headers=headers, timeout=self.http_request_timeout
-            response.raise_for_status()
+            )
-            email_info = response.json()
+            email_response.raise_for_status()
-            user_info["email"] = next(
+            email_info = email_response.json()
-                (email for email in email_info if email["primary"]), None
+            user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"]
-            )["email"]
             return self.normalize_user_info(user_info)
-        except requests.exceptions.RequestException as e:
+        except Exception as e:
+            raise ValueError(f"Failed to fetch github user info: {e}")
+
+    async def async_fetch_user_info(self, access_token, **kwargs):
+        """Async variant of fetch_user_info using httpx."""
+        user_info = {}
+        headers = {"Authorization": f"Bearer {access_token}"}
+        try:
+            response = await async_request(
+                "GET",
+                self.userinfo_url,
+                headers=headers,
+                timeout=self.http_request_timeout,
+            )
+            response.raise_for_status()
+            user_info.update(response.json())
+
+            email_response = await async_request(
+                "GET",
+                self.userinfo_url + "/emails",
+                headers=headers,
+                timeout=self.http_request_timeout,
+            )
+            email_response.raise_for_status()
+            email_info = email_response.json()
+            user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"]
+            return self.normalize_user_info(user_info)
+        except Exception as e:
             raise ValueError(f"Failed to fetch github user info: {e}")
 
 

api/apps/auth/oauth.py

@@ -14,8 +14,8 @@
 #  limitations under the License.
 #
 
-import requests
 import urllib.parse
+from common.http_client import async_request, sync_request
 
 
 class UserInfo:
@@ -74,15 +74,40 @@ class OAuthClient:
                 "redirect_uri": self.redirect_uri,
                 "grant_type": "authorization_code"
             }
-            response = requests.post(
+            response = sync_request(
+                "POST",
                 self.token_url,
                 data=payload,
                 headers={"Accept": "application/json"},
-                timeout=self.http_request_timeout
+                timeout=self.http_request_timeout,
             )
             response.raise_for_status()
             return response.json()
-        except requests.exceptions.RequestException as e:
+        except Exception as e:
+            raise ValueError(f"Failed to exchange authorization code for token: {e}")
+
+    async def async_exchange_code_for_token(self, code):
+        """
+        Async variant of exchange_code_for_token using httpx.
+        """
+        payload = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "code": code,
+            "redirect_uri": self.redirect_uri,
+            "grant_type": "authorization_code",
+        }
+        try:
+            response = await async_request(
+                "POST",
+                self.token_url,
+                data=payload,
+                headers={"Accept": "application/json"},
+                timeout=self.http_request_timeout,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
             raise ValueError(f"Failed to exchange authorization code for token: {e}")
 
 
@@ -92,11 +117,27 @@ class OAuthClient:
         """
         try:
             headers = {"Authorization": f"Bearer {access_token}"}
-            response = requests.get(self.userinfo_url, headers=headers, timeout=self.http_request_timeout)
+            response = sync_request("GET", self.userinfo_url, headers=headers, timeout=self.http_request_timeout)
             response.raise_for_status()
             user_info = response.json()
             return self.normalize_user_info(user_info)
-        except requests.exceptions.RequestException as e:
+        except Exception as e:
+            raise ValueError(f"Failed to fetch user info: {e}")
+
+    async def async_fetch_user_info(self, access_token, **kwargs):
+        """Async variant of fetch_user_info using httpx."""
+        headers = {"Authorization": f"Bearer {access_token}"}
+        try:
+            response = await async_request(
+                "GET",
+                self.userinfo_url,
+                headers=headers,
+                timeout=self.http_request_timeout,
+            )
+            response.raise_for_status()
+            user_info = response.json()
+            return self.normalize_user_info(user_info)
+        except Exception as e:
             raise ValueError(f"Failed to fetch user info: {e}")
 
 

api/apps/auth/oidc.py

@@ -15,7 +15,7 @@
 #
 
 import jwt
-import requests
+from common.http_client import sync_request
 from .oauth import OAuthClient
 
 
@@ -50,10 +50,10 @@ class OIDCClient(OAuthClient):
         """
         try:
             metadata_url = f"{issuer}/.well-known/openid-configuration"
-            response = requests.get(metadata_url, timeout=7)
+            response = sync_request("GET", metadata_url, timeout=7)
             response.raise_for_status()
             return response.json()
-        except requests.exceptions.RequestException as e:
+        except Exception as e:
             raise ValueError(f"Failed to fetch OIDC metadata: {e}")
 
 
@@ -95,6 +95,13 @@ class OIDCClient(OAuthClient):
         user_info.update(super().fetch_user_info(access_token).to_dict())
         return self.normalize_user_info(user_info)
 
+    async def async_fetch_user_info(self, access_token, id_token=None, **kwargs):
+        user_info = {}
+        if id_token:
+            user_info = self.parse_id_token(id_token)
+        user_info.update((await super().async_fetch_user_info(access_token)).to_dict())
+        return self.normalize_user_info(user_info)
+
 
     def normalize_user_info(self, user_info):
         return super().normalize_user_info(user_info)

api/apps/connector_app.py

@@ -26,7 +26,7 @@ from google_auth_oauthlib.flow import Flow
 
 from api.db import InputType
 from api.db.services.connector_service import ConnectorService, SyncLogsService
-from api.utils.api_utils import get_data_error_result, get_json_result, validate_request
+from api.utils.api_utils import get_data_error_result, get_json_result, request_json, validate_request
 from common.constants import RetCode, TaskStatus
 from common.data_source.config import GOOGLE_DRIVE_WEB_OAUTH_REDIRECT_URI, DocumentSource
 from common.data_source.google_util.constant import GOOGLE_DRIVE_WEB_OAUTH_POPUP_TEMPLATE, GOOGLE_SCOPES
@@ -38,7 +38,7 @@ from api.apps import login_required, current_user
 @manager.route("/set", methods=["POST"])  # noqa: F821
 @login_required
 async def set_connector():
-    req = await request.json
+    req = await request_json()
     if req.get("id"):
         conn = {fld: req[fld] for fld in ["prune_freq", "refresh_freq", "config", "timeout_secs"] if fld in req}
         ConnectorService.update_by_id(req["id"], conn)
@@ -90,7 +90,7 @@ def list_logs(connector_id):
 @manager.route("/<connector_id>/resume", methods=["PUT"])  # noqa: F821
 @login_required
 async def resume(connector_id):
-    req = await request.json
+    req = await request_json()
     if req.get("resume"):
         ConnectorService.resume(connector_id, TaskStatus.SCHEDULE)
     else:
@@ -102,7 +102,7 @@ async def resume(connector_id):
 @login_required
 @validate_request("kb_id")
 async def rebuild(connector_id):
-    req = await request.json
+    req = await request_json()
     err = ConnectorService.rebuild(req["kb_id"], connector_id, current_user.id)
     if err:
         return get_json_result(data=False, message=err, code=RetCode.SERVER_ERROR)
@@ -179,7 +179,7 @@ async def start_google_drive_web_oauth():
             message="Google Drive OAuth redirect URI is not configured on the server.",
         )
 
-    req = await request.json or {}
+    req = await request_json()
     raw_credentials = req.get("credentials", "")
     try:
         credentials = _load_credentials(raw_credentials)
@@ -281,7 +281,7 @@ async def google_drive_web_oauth_callback():
 @login_required
 @validate_request("flow_id")
 async def poll_google_drive_web_result():
-    req = await request.json or {}
+    req = await request_json()
     flow_id = req.get("flow_id")
     cache_raw = REDIS_CONN.get(_web_result_cache_key(flow_id))
     if not cache_raw:

api/apps/conversation_app.py

@@ -26,7 +26,7 @@ from api.db.services.llm_service import LLMBundle
 from api.db.services.search_service import SearchService
 from api.db.services.tenant_llm_service import TenantLLMService
 from api.db.services.user_service import TenantService, UserTenantService
-from api.utils.api_utils import get_data_error_result, get_json_result, server_error_response, validate_request
+from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request
 from rag.prompts.template import load_prompt
 from rag.prompts.generator import chunks_format
 from common.constants import RetCode, LLMType
@@ -35,7 +35,7 @@ from common.constants import RetCode, LLMType
 @manager.route("/set", methods=["POST"])  # noqa: F821
 @login_required
 async def set_conversation():
-    req = await request.json
+    req = await request_json()
     conv_id = req.get("conversation_id")
     is_new = req.get("is_new")
     name = req.get("name", "New conversation")
@@ -78,7 +78,7 @@ async def set_conversation():
 
 @manager.route("/get", methods=["GET"])  # noqa: F821
 @login_required
-def get():
+async def get():
     conv_id = request.args["conversation_id"]
     try:
         e, conv = ConversationService.get_by_id(conv_id)
@@ -129,7 +129,7 @@ def getsse(dialog_id):
 @manager.route("/rm", methods=["POST"])  # noqa: F821
 @login_required
 async def rm():
-    req = await request.json
+    req = await request_json()
     conv_ids = req["conversation_ids"]
     try:
         for cid in conv_ids:
@@ -150,7 +150,7 @@ async def rm():
 
 @manager.route("/list", methods=["GET"])  # noqa: F821
 @login_required
-def list_conversation():
+async def list_conversation():
     dialog_id = request.args["dialog_id"]
     try:
         if not DialogService.query(tenant_id=current_user.id, id=dialog_id):
@@ -167,7 +167,7 @@ def list_conversation():
 @login_required
 @validate_request("conversation_id", "messages")
 async def completion():
-    req = await request.json
+    req = await request_json()
     msg = []
     for m in req["messages"]:
         if m["role"] == "system":
@@ -252,7 +252,7 @@ async def completion():
 @manager.route("/tts", methods=["POST"])  # noqa: F821
 @login_required
 async def tts():
-    req = await request.json
+    req = await request_json()
     text = req["text"]
 
     tenants = TenantService.get_info_by(current_user.id)
@@ -285,7 +285,7 @@ async def tts():
 @login_required
 @validate_request("conversation_id", "message_id")
 async def delete_msg():
-    req = await request.json
+    req = await request_json()
     e, conv = ConversationService.get_by_id(req["conversation_id"])
     if not e:
         return get_data_error_result(message="Conversation not found!")
@@ -308,7 +308,7 @@ async def delete_msg():
 @login_required
 @validate_request("conversation_id", "message_id")
 async def thumbup():
-    req = await request.json
+    req = await request_json()
     e, conv = ConversationService.get_by_id(req["conversation_id"])
     if not e:
         return get_data_error_result(message="Conversation not found!")
@@ -335,7 +335,7 @@ async def thumbup():
 @login_required
 @validate_request("question", "kb_ids")
 async def ask_about():
-    req = await request.json
+    req = await request_json()
     uid = current_user.id
 
     search_id = req.get("search_id", "")
@@ -367,7 +367,7 @@ async def ask_about():
 @login_required
 @validate_request("question", "kb_ids")
 async def mindmap():
-    req = await request.json
+    req = await request_json()
     search_id = req.get("search_id", "")
     search_app = SearchService.get_detail(search_id) if search_id else {}
     search_config = search_app.get("search_config", {}) if search_app else {}
@@ -385,7 +385,7 @@ async def mindmap():
 @login_required
 @validate_request("question")
 async def related_questions():
-    req = await request.json
+    req = await request_json()
 
     search_id = req.get("search_id", "")
     search_config = {}

api/apps/dialog_app.py

@@ -21,10 +21,9 @@ from common.constants import StatusEnum
 from api.db.services.tenant_llm_service import TenantLLMService
 from api.db.services.knowledgebase_service import KnowledgebaseService
 from api.db.services.user_service import TenantService, UserTenantService
-from api.utils.api_utils import server_error_response, get_data_error_result, validate_request
+from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request
 from common.misc_utils import get_uuid
 from common.constants import RetCode
-from api.utils.api_utils import get_json_result
 from api.apps import login_required, current_user
 
 
@@ -32,7 +31,7 @@ from api.apps import login_required, current_user
 @validate_request("prompt_config")
 @login_required
 async def set_dialog():
-    req = await request.json
+    req = await request_json()
     dialog_id = req.get("dialog_id", "")
     is_create = not dialog_id
     name = req.get("name", "New Dialog")
@@ -181,7 +180,7 @@ async def list_dialogs_next():
     else:
         desc = True
 
-    req = await request.get_json()
+    req = await request_json()
     owner_ids = req.get("owner_ids", [])
     try:
         if not owner_ids:
@@ -209,7 +208,7 @@ async def list_dialogs_next():
 @login_required
 @validate_request("dialog_ids")
 async def rm():
-    req = await request.json
+    req = await request_json()
     dialog_list=[]
     tenants = UserTenantService.query(user_id=current_user.id)
     try:

api/apps/document_app.py

@@ -230,7 +230,7 @@ async def list_docs():
     create_time_from = int(request.args.get("create_time_from", 0))
     create_time_to = int(request.args.get("create_time_to", 0))
 
-    req = await request.get_json()
+    req = await request_json()
 
     run_status = req.get("run_status", [])
     if run_status:
@@ -271,7 +271,7 @@ async def list_docs():
 @manager.route("/filter", methods=["POST"])  # noqa: F821
 @login_required
 async def get_filter():
-    req = await request.get_json()
+    req = await request_json()
 
     kb_id = req.get("kb_id")
     if not kb_id:
@@ -341,7 +341,7 @@ def thumbnails():
 @login_required
 @validate_request("doc_ids", "status")
 async def change_status():
-    req = await request.get_json()
+    req = await request_json()
     doc_ids = req.get("doc_ids", [])
     status = str(req.get("status", ""))
 
@@ -624,7 +624,8 @@ async def upload_and_parse():
 @manager.route("/parse", methods=["POST"])  # noqa: F821
 @login_required
 async def parse():
-    url = await request.json.get("url") if await request.json else ""
+    req = await request_json()
+    url = req.get("url", "")
     if url:
         if not is_valid_url(url):
             return get_json_result(data=False, message="The URL format is invalid", code=RetCode.ARGUMENT_ERROR)

api/apps/file2document_app.py

@@ -19,22 +19,20 @@ from pathlib import Path
 from api.db.services.file2document_service import File2DocumentService
 from api.db.services.file_service import FileService
 
-from quart import request
 from api.apps import login_required, current_user
 from api.db.services.knowledgebase_service import KnowledgebaseService
-from api.utils.api_utils import server_error_response, get_data_error_result, validate_request
+from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request
 from common.misc_utils import get_uuid
 from common.constants import RetCode
 from api.db import FileType
 from api.db.services.document_service import DocumentService
-from api.utils.api_utils import get_json_result
 
 
 @manager.route('/convert', methods=['POST'])  # noqa: F821
 @login_required
 @validate_request("file_ids", "kb_ids")
 async def convert():
-    req = await request.json
+    req = await request_json()
     kb_ids = req["kb_ids"]
     file_ids = req["file_ids"]
     file2documents = []
@@ -104,7 +102,7 @@ async def convert():
 @login_required
 @validate_request("file_ids")
 async def rm():
-    req = await request.json
+    req = await request_json()
     file_ids = req["file_ids"]
     if not file_ids:
         return get_json_result(

api/apps/file_app.py

@@ -29,7 +29,7 @@ from common.constants import RetCode, FileSource
 from api.db import FileType
 from api.db.services import duplicate_name
 from api.db.services.file_service import FileService
-from api.utils.api_utils import get_json_result
+from api.utils.api_utils import get_json_result, request_json
 from api.utils.file_utils import filename_type
 from api.utils.web_utils import CONTENT_TYPE_MAP
 from common import settings
@@ -124,7 +124,7 @@ async def upload():
 @login_required
 @validate_request("name")
 async def create():
-    req = await request.json
+    req = await request_json()
     pf_id = req.get("parent_id")
     input_file_type = req.get("type")
     if not pf_id:
@@ -239,7 +239,7 @@ def get_all_parent_folders():
 @login_required
 @validate_request("file_ids")
 async def rm():
-    req = await request.json
+    req = await request_json()
     file_ids = req["file_ids"]
 
     def _delete_single_file(file):
@@ -300,7 +300,7 @@ async def rm():
 @login_required
 @validate_request("file_id", "name")
 async def rename():
-    req = await request.json
+    req = await request_json()
     try:
         e, file = FileService.get_by_id(req["file_id"])
         if not e:
@@ -369,7 +369,7 @@ async def get(file_id):
 @login_required
 @validate_request("src_file_ids", "dest_file_id")
 async def move():
-    req = await request.json
+    req = await request_json()
     try:
         file_ids = req["src_file_ids"]
         dest_parent_id = req["dest_file_id"]

api/apps/langfuse_app.py

@@ -15,20 +15,19 @@
 #
 
 
-from quart import request
 from api.apps import current_user, login_required
 from langfuse import Langfuse
 
 from api.db.db_models import DB
 from api.db.services.langfuse_service import TenantLangfuseService
-from api.utils.api_utils import get_error_data_result, get_json_result, server_error_response, validate_request
+from api.utils.api_utils import get_error_data_result, get_json_result, request_json, server_error_response, validate_request
 
 
 @manager.route("/api_key", methods=["POST", "PUT"])  # noqa: F821
 @login_required
 @validate_request("secret_key", "public_key", "host")
 async def set_api_key():
-    req = await request.get_json()
+    req = await request_json()
     secret_key = req.get("secret_key", "")
     public_key = req.get("public_key", "")
     host = req.get("host", "")

api/apps/llm_app.py

@@ -21,10 +21,9 @@ from quart import request
 from api.apps import login_required, current_user
 from api.db.services.tenant_llm_service import LLMFactoriesService, TenantLLMService
 from api.db.services.llm_service import LLMService
-from api.utils.api_utils import server_error_response, get_data_error_result, validate_request
+from api.utils.api_utils import get_allowed_llm_factories, get_data_error_result, get_json_result, request_json, server_error_response, validate_request
 from common.constants import StatusEnum, LLMType
 from api.db.db_models import TenantLLM
-from api.utils.api_utils import get_json_result, get_allowed_llm_factories
 from rag.utils.base64_image import test_image
 from rag.llm import EmbeddingModel, ChatModel, RerankModel, CvModel, TTSModel
 
@@ -54,7 +53,7 @@ def factories():
 @login_required
 @validate_request("llm_factory", "api_key")
 async def set_api_key():
-    req = await request.json
+    req = await request_json()
     # test if api key works
     chat_passed, embd_passed, rerank_passed = False, False, False
     factory = req["llm_factory"]
@@ -124,7 +123,7 @@ async def set_api_key():
 @login_required
 @validate_request("llm_factory")
 async def add_llm():
-    req = await request.json
+    req = await request_json()
     factory = req["llm_factory"]
     api_key = req.get("api_key", "x")
     llm_name = req.get("llm_name")
@@ -269,7 +268,7 @@ async def add_llm():
 @login_required
 @validate_request("llm_factory", "llm_name")
 async def delete_llm():
-    req = await request.json
+    req = await request_json()
     TenantLLMService.filter_delete([TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"], TenantLLM.llm_name == req["llm_name"]])
     return get_json_result(data=True)
 
@@ -278,7 +277,7 @@ async def delete_llm():
 @login_required
 @validate_request("llm_factory", "llm_name")
 async def enable_llm():
-    req = await request.json
+    req = await request_json()
     TenantLLMService.filter_update(
         [TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"], TenantLLM.llm_name == req["llm_name"]], {"status": str(req.get("status", "1"))}
     )
@@ -289,7 +288,7 @@ async def enable_llm():
 @login_required
 @validate_request("llm_factory")
 async def delete_factory():
-    req = await request.json
+    req = await request_json()
     TenantLLMService.filter_delete([TenantLLM.tenant_id == current_user.id, TenantLLM.llm_factory == req["llm_factory"]])
     return get_json_result(data=True)
 

api/apps/mcp_server_app.py

@@ -22,8 +22,7 @@ from api.db.services.user_service import TenantService
 from common.constants import RetCode, VALID_MCP_SERVER_TYPES
 
 from common.misc_utils import get_uuid
-from api.utils.api_utils import get_data_error_result, get_json_result, server_error_response, validate_request, \
+from api.utils.api_utils import get_data_error_result, get_json_result, get_mcp_tools, request_json, server_error_response, validate_request
-    get_mcp_tools
 from api.utils.web_utils import get_float, safe_json_parse
 from common.mcp_tool_call_conn import MCPToolCallSession, close_multiple_mcp_toolcall_sessions
 
@@ -40,7 +39,7 @@ async def list_mcp() -> Response:
     else:
         desc = True
 
-    req = await request.get_json()
+    req = await request_json()
     mcp_ids = req.get("mcp_ids", [])
     try:
         servers = MCPServerService.get_servers(current_user.id, mcp_ids, 0, 0, orderby, desc, keywords) or []
@@ -73,7 +72,7 @@ def detail() -> Response:
 @login_required
 @validate_request("name", "url", "server_type")
 async def create() -> Response:
-    req = await request.get_json()
+    req = await request_json()
 
     server_type = req.get("server_type", "")
     if server_type not in VALID_MCP_SERVER_TYPES:
@@ -128,7 +127,7 @@ async def create() -> Response:
 @login_required
 @validate_request("mcp_id")
 async def update() -> Response:
-    req = await request.get_json()
+    req = await request_json()
 
     mcp_id = req.get("mcp_id", "")
     e, mcp_server = MCPServerService.get_by_id(mcp_id)
@@ -184,7 +183,7 @@ async def update() -> Response:
 @login_required
 @validate_request("mcp_ids")
 async def rm() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     mcp_ids = req.get("mcp_ids", [])
 
     try:
@@ -202,7 +201,7 @@ async def rm() -> Response:
 @login_required
 @validate_request("mcpServers")
 async def import_multiple() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     servers = req.get("mcpServers", {})
     if not servers:
         return get_data_error_result(message="No MCP servers provided.")
@@ -269,7 +268,7 @@ async def import_multiple() -> Response:
 @login_required
 @validate_request("mcp_ids")
 async def export_multiple() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     mcp_ids = req.get("mcp_ids", [])
 
     if not mcp_ids:
@@ -301,7 +300,7 @@ async def export_multiple() -> Response:
 @login_required
 @validate_request("mcp_ids")
 async def list_tools() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     mcp_ids = req.get("mcp_ids", [])
     if not mcp_ids:
         return get_data_error_result(message="No MCP server IDs provided.")
@@ -348,7 +347,7 @@ async def list_tools() -> Response:
 @login_required
 @validate_request("mcp_id", "tool_name", "arguments")
 async def test_tool() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     mcp_id = req.get("mcp_id", "")
     if not mcp_id:
         return get_data_error_result(message="No MCP server ID provided.")
@@ -381,7 +380,7 @@ async def test_tool() -> Response:
 @login_required
 @validate_request("mcp_id", "tools")
 async def cache_tool() -> Response:
-    req = await request.get_json()
+    req = await request_json()
     mcp_id = req.get("mcp_id", "")
     if not mcp_id:
         return get_data_error_result(message="No MCP server ID provided.")
@@ -404,7 +403,7 @@ async def cache_tool() -> Response:
 @manager.route("/test_mcp", methods=["POST"])  # noqa: F821
 @validate_request("url", "server_type")
 async def test_mcp() -> Response:
-    req = await request.get_json()
+    req = await request_json()
 
     url = req.get("url", "")
     if not url:

api/apps/sdk/agents.py

@@ -25,7 +25,7 @@ from api.db.services.canvas_service import UserCanvasService
 from api.db.services.user_canvas_version import UserCanvasVersionService
 from common.constants import RetCode
 from common.misc_utils import get_uuid
-from api.utils.api_utils import get_data_error_result, get_error_data_result, get_json_result, token_required
+from api.utils.api_utils import get_data_error_result, get_error_data_result, get_json_result, request_json, token_required
 from api.utils.api_utils import get_result
 from quart import request, Response
 
@@ -53,7 +53,7 @@ def list_agents(tenant_id):
 @manager.route("/agents", methods=["POST"])  # noqa: F821
 @token_required
 async def create_agent(tenant_id: str):
-    req: dict[str, Any] = cast(dict[str, Any], await request.json)
+    req: dict[str, Any] = cast(dict[str, Any], await request_json())
     req["user_id"] = tenant_id
 
     if req.get("dsl") is not None:
@@ -90,7 +90,7 @@ async def create_agent(tenant_id: str):
 @manager.route("/agents/<agent_id>", methods=["PUT"])  # noqa: F821
 @token_required
 async def update_agent(tenant_id: str, agent_id: str):
-    req: dict[str, Any] = {k: v for k, v in cast(dict[str, Any], (await request.json)).items() if v is not None}
+    req: dict[str, Any] = {k: v for k, v in cast(dict[str, Any], (await request_json())).items() if v is not None}
     req["user_id"] = tenant_id
 
     if req.get("dsl") is not None:
@@ -136,7 +136,7 @@ def delete_agent(tenant_id: str, agent_id: str):
 @manager.route('/webhook/<agent_id>', methods=['POST'])  # noqa: F821
 @token_required
 async def webhook(tenant_id: str, agent_id: str):
-    req = await request.json
+    req = await request_json()
     if not UserCanvasService.accessible(req["id"], tenant_id):
         return get_json_result(
             data=False, message='Only owner of canvas authorized for this operation.',

api/apps/sdk/dify_retrieval.py

@@ -15,12 +15,12 @@
 #
 import logging
 
-from quart import request, jsonify
+from quart import jsonify
 
 from api.db.services.document_service import DocumentService
 from api.db.services.knowledgebase_service import KnowledgebaseService
 from api.db.services.llm_service import LLMBundle
-from api.utils.api_utils import validate_request, build_error_result, apikey_required
+from api.utils.api_utils import apikey_required, build_error_result, request_json, validate_request
 from rag.app.tag import label_question
 from api.db.services.dialog_service import meta_filter, convert_conditions
 from common.constants import RetCode, LLMType
@@ -113,7 +113,7 @@ async def retrieval(tenant_id):
       404:
         description: Knowledge base or document not found
     """
-    req = await request.json
+    req = await request_json()
     question = req["query"]
     kb_id = req["knowledge_id"]
     use_kg = req.get("use_kg", False)

api/apps/sdk/files.py

@@ -23,12 +23,11 @@ from pathlib import Path
 from api.db.services.document_service import DocumentService
 from api.db.services.file2document_service import File2DocumentService
 from api.db.services.knowledgebase_service import KnowledgebaseService
-from api.utils.api_utils import server_error_response, token_required
+from api.utils.api_utils import get_json_result, request_json, server_error_response, token_required
 from common.misc_utils import get_uuid
 from api.db import FileType
 from api.db.services import duplicate_name
 from api.db.services.file_service import FileService
-from api.utils.api_utils import get_json_result
 from api.utils.file_utils import filename_type
 from common import settings
 from common.constants import RetCode
@@ -193,9 +192,9 @@ async def create(tenant_id):
                 type:
                   type: string
     """
-    req = await request.json
+    req = await request_json()
-    pf_id = await request.json.get("parent_id")
+    pf_id = req.get("parent_id")
-    input_file_type = await request.json.get("type")
+    input_file_type = req.get("type")
     if not pf_id:
         root_folder = FileService.get_root_folder(tenant_id)
         pf_id = root_folder["id"]
@@ -229,7 +228,7 @@ async def create(tenant_id):
 
 @manager.route('/file/list', methods=['GET'])  # noqa: F821
 @token_required
-def list_files(tenant_id):
+async def list_files(tenant_id):
     """
     List files under a specific folder.
     ---
@@ -321,7 +320,7 @@ def list_files(tenant_id):
 
 @manager.route('/file/root_folder', methods=['GET'])  # noqa: F821
 @token_required
-def get_root_folder(tenant_id):
+async def get_root_folder(tenant_id):
     """
     Get user's root folder.
     ---
@@ -357,7 +356,7 @@ def get_root_folder(tenant_id):
 
 @manager.route('/file/parent_folder', methods=['GET'])  # noqa: F821
 @token_required
-def get_parent_folder():
+async def get_parent_folder():
     """
     Get parent folder info of a file.
     ---
@@ -402,7 +401,7 @@ def get_parent_folder():
 
 @manager.route('/file/all_parent_folder', methods=['GET'])  # noqa: F821
 @token_required
-def get_all_parent_folders(tenant_id):
+async def get_all_parent_folders(tenant_id):
     """
     Get all parent folders of a file.
     ---
@@ -481,7 +480,7 @@ async def rm(tenant_id):
               type: boolean
               example: true
     """
-    req = await request.json
+    req = await request_json()
     file_ids = req["file_ids"]
     try:
         for file_id in file_ids:
@@ -556,7 +555,7 @@ async def rename(tenant_id):
               type: boolean
               example: true
     """
-    req = await request.json
+    req = await request_json()
     try:
         e, file = FileService.get_by_id(req["file_id"])
         if not e:
@@ -667,7 +666,7 @@ async def move(tenant_id):
               type: boolean
               example: true
     """
-    req = await request.json
+    req = await request_json()
     try:
         file_ids = req["src_file_ids"]
         parent_id = req["dest_file_id"]
@@ -694,7 +693,7 @@ async def move(tenant_id):
 @manager.route('/file/convert', methods=['POST'])  # noqa: F821
 @token_required
 async def convert(tenant_id):
-    req = await request.json
+    req = await request_json()
     kb_ids = req["kb_ids"]
     file_ids = req["file_ids"]
     file2documents = []

api/apps/sdk/session.py

@@ -35,7 +35,7 @@ from api.db.services.search_service import SearchService
 from api.db.services.user_service import UserTenantService
 from common.misc_utils import get_uuid
 from api.utils.api_utils import check_duplicate_ids, get_data_openai, get_error_data_result, get_json_result, \
-    get_result, server_error_response, token_required, validate_request
+    get_result, request_json, server_error_response, token_required, validate_request
 from rag.app.tag import label_question
 from rag.prompts.template import load_prompt
 from rag.prompts.generator import cross_languages, gen_meta_filter, keyword_extraction, chunks_format
@@ -45,7 +45,7 @@ from common import settings
 @manager.route("/chats/<chat_id>/sessions", methods=["POST"])  # noqa: F821
 @token_required
 async def create(tenant_id, chat_id):
-    req = await request.json
+    req = await request_json()
     req["dialog_id"] = chat_id
     dia = DialogService.query(tenant_id=tenant_id, id=req["dialog_id"], status=StatusEnum.VALID.value)
     if not dia:
@@ -73,7 +73,7 @@ async def create(tenant_id, chat_id):
 
 @manager.route("/agents/<agent_id>/sessions", methods=["POST"])  # noqa: F821
 @token_required
-def create_agent_session(tenant_id, agent_id):
+async def create_agent_session(tenant_id, agent_id):
     user_id = request.args.get("user_id", tenant_id)
     e, cvs = UserCanvasService.get_by_id(agent_id)
     if not e:
@@ -98,7 +98,7 @@ def create_agent_session(tenant_id, agent_id):
 @manager.route("/chats/<chat_id>/sessions/<session_id>", methods=["PUT"])  # noqa: F821
 @token_required
 async def update(tenant_id, chat_id, session_id):
-    req = await request.json
+    req = await request_json()
     req["dialog_id"] = chat_id
     conv_id = session_id
     conv = ConversationService.query(id=conv_id, dialog_id=chat_id)
@@ -120,7 +120,7 @@ async def update(tenant_id, chat_id, session_id):
 @manager.route("/chats/<chat_id>/completions", methods=["POST"])  # noqa: F821
 @token_required
 async def chat_completion(tenant_id, chat_id):
-    req = await request.json
+    req = await request_json()
     if not req:
         req = {"question": ""}
     if not req.get("session_id"):
@@ -206,7 +206,7 @@ async def chat_completion_openai_like(tenant_id, chat_id):
         if reference:
             print(completion.choices[0].message.reference)
     """
-    req = await request.get_json()
+    req = await request_json()
 
     need_reference = bool(req.get("reference", False))
 
@@ -384,7 +384,7 @@ async def chat_completion_openai_like(tenant_id, chat_id):
 @validate_request("model", "messages")  # noqa: F821
 @token_required
 async def agents_completion_openai_compatibility(tenant_id, agent_id):
-    req = await request.json
+    req = await request_json()
     tiktokenenc = tiktoken.get_encoding("cl100k_base")
     messages = req.get("messages", [])
     if not messages:
@@ -442,7 +442,7 @@ async def agents_completion_openai_compatibility(tenant_id, agent_id):
 @manager.route("/agents/<agent_id>/completions", methods=["POST"])  # noqa: F821
 @token_required
 async def agent_completions(tenant_id, agent_id):
-    req = await request.json
+    req = await request_json()
 
     if req.get("stream", True):
 
@@ -491,7 +491,7 @@ async def agent_completions(tenant_id, agent_id):
 
 @manager.route("/chats/<chat_id>/sessions", methods=["GET"])  # noqa: F821
 @token_required
-def list_session(tenant_id, chat_id):
+async def list_session(tenant_id, chat_id):
     if not DialogService.query(tenant_id=tenant_id, id=chat_id, status=StatusEnum.VALID.value):
         return get_error_data_result(message=f"You don't own the assistant {chat_id}.")
     id = request.args.get("id")
@@ -545,7 +545,7 @@ def list_session(tenant_id, chat_id):
 
 @manager.route("/agents/<agent_id>/sessions", methods=["GET"])  # noqa: F821
 @token_required
-def list_agent_session(tenant_id, agent_id):
+async def list_agent_session(tenant_id, agent_id):
     if not UserCanvasService.query(user_id=tenant_id, id=agent_id):
         return get_error_data_result(message=f"You don't own the agent {agent_id}.")
     id = request.args.get("id")
@@ -614,7 +614,7 @@ async def delete(tenant_id, chat_id):
 
     errors = []
     success_count = 0
-    req = await request.json
+    req = await request_json()
     convs = ConversationService.query(dialog_id=chat_id)
     if not req:
         ids = None
@@ -662,7 +662,7 @@ async def delete(tenant_id, chat_id):
 async def delete_agent_session(tenant_id, agent_id):
     errors = []
     success_count = 0
-    req = await request.json
+    req = await request_json()
     cvs = UserCanvasService.query(user_id=tenant_id, id=agent_id)
     if not cvs:
         return get_error_data_result(f"You don't own the agent {agent_id}")
@@ -715,7 +715,7 @@ async def delete_agent_session(tenant_id, agent_id):
 @manager.route("/sessions/ask", methods=["POST"])  # noqa: F821
 @token_required
 async def ask_about(tenant_id):
-    req = await request.json
+    req = await request_json()
     if not req.get("question"):
         return get_error_data_result("`question` is required.")
     if not req.get("dataset_ids"):
@@ -754,7 +754,7 @@ async def ask_about(tenant_id):
 @manager.route("/sessions/related_questions", methods=["POST"])  # noqa: F821
 @token_required
 async def related_questions(tenant_id):
-    req = await request.json
+    req = await request_json()
     if not req.get("question"):
         return get_error_data_result("`question` is required.")
     question = req["question"]
@@ -805,7 +805,7 @@ Related search terms:
 
 @manager.route("/chatbots/<dialog_id>/completions", methods=["POST"])  # noqa: F821
 async def chatbot_completions(dialog_id):
-    req = await request.json
+    req = await request_json()
 
     token = request.headers.get("Authorization").split()
     if len(token) != 2:
@@ -831,7 +831,7 @@ async def chatbot_completions(dialog_id):
 
 
 @manager.route("/chatbots/<dialog_id>/info", methods=["GET"])  # noqa: F821
-def chatbots_inputs(dialog_id):
+async def chatbots_inputs(dialog_id):
     token = request.headers.get("Authorization").split()
     if len(token) != 2:
         return get_error_data_result(message='Authorization is not valid!"')
@@ -855,7 +855,7 @@ def chatbots_inputs(dialog_id):
 
 @manager.route("/agentbots/<agent_id>/completions", methods=["POST"])  # noqa: F821
 async def agent_bot_completions(agent_id):
-    req = await request.json
+    req = await request_json()
 
     token = request.headers.get("Authorization").split()
     if len(token) != 2:
@@ -878,7 +878,7 @@ async def agent_bot_completions(agent_id):
 
 
 @manager.route("/agentbots/<agent_id>/inputs", methods=["GET"])  # noqa: F821
-def begin_inputs(agent_id):
+async def begin_inputs(agent_id):
     token = request.headers.get("Authorization").split()
     if len(token) != 2:
         return get_error_data_result(message='Authorization is not valid!"')
@@ -908,7 +908,7 @@ async def ask_about_embedded():
     if not objs:
         return get_error_data_result(message='Authentication error: API key is invalid!"')
 
-    req = await request.json
+    req = await request_json()
     uid = objs[0].tenant_id
 
     search_id = req.get("search_id", "")
@@ -947,7 +947,7 @@ async def retrieval_test_embedded():
     if not objs:
         return get_error_data_result(message='Authentication error: API key is invalid!"')
 
-    req = await request.json
+    req = await request_json()
     page = int(req.get("page", 1))
     size = int(req.get("size", 30))
     question = req["question"]
@@ -1046,7 +1046,7 @@ async def related_questions_embedded():
     if not objs:
         return get_error_data_result(message='Authentication error: API key is invalid!"')
 
-    req = await request.json
+    req = await request_json()
     tenant_id = objs[0].tenant_id
     if not tenant_id:
         return get_error_data_result(message="permission denined.")
@@ -1081,7 +1081,7 @@ Related search terms:
 
 
 @manager.route("/searchbots/detail", methods=["GET"])  # noqa: F821
-def detail_share_embedded():
+async def detail_share_embedded():
     token = request.headers.get("Authorization").split()
     if len(token) != 2:
         return get_error_data_result(message='Authorization is not valid!"')
@@ -1123,7 +1123,7 @@ async def mindmap():
         return get_error_data_result(message='Authentication error: API key is invalid!"')
 
     tenant_id = objs[0].tenant_id
-    req = await request.json
+    req = await request_json()
 
     search_id = req.get("search_id", "")
     search_app = SearchService.get_detail(search_id) if search_id else {}

api/apps/search_app.py

@@ -24,14 +24,14 @@ from api.db.services.search_service import SearchService
 from api.db.services.user_service import TenantService, UserTenantService
 from common.misc_utils import get_uuid
 from common.constants import RetCode, StatusEnum
-from api.utils.api_utils import get_data_error_result, get_json_result, not_allowed_parameters, server_error_response, validate_request
+from api.utils.api_utils import get_data_error_result, get_json_result, not_allowed_parameters, request_json, server_error_response, validate_request
 
 
 @manager.route("/create", methods=["post"])  # noqa: F821
 @login_required
 @validate_request("name")
 async def create():
-    req = await request.get_json()
+    req = await request_json()
     search_name = req["name"]
     description = req.get("description", "")
     if not isinstance(search_name, str):
@@ -66,7 +66,7 @@ async def create():
 @validate_request("search_id", "name", "search_config", "tenant_id")
 @not_allowed_parameters("id", "created_by", "create_time", "update_time", "create_date", "update_date", "created_by")
 async def update():
-    req = await request.get_json()
+    req = await request_json()
     if not isinstance(req["name"], str):
         return get_data_error_result(message="Search name must be string.")
     if req["name"].strip() == "":
@@ -150,7 +150,7 @@ async def list_search_app():
     else:
         desc = True
 
-    req = await request.get_json()
+    req = await request_json()
     owner_ids = req.get("owner_ids", [])
     try:
         if not owner_ids:
@@ -174,7 +174,7 @@ async def list_search_app():
 @login_required
 @validate_request("search_id")
 async def rm():
-    req = await request.get_json()
+    req = await request_json()
     search_id = req["search_id"]
     if not SearchService.accessible4deletion(search_id, current_user.id):
         return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR)

api/apps/tenant_app.py

@@ -14,7 +14,6 @@
 #  limitations under the License.
 #
 
-from quart import request
 from api.db import UserTenantRole
 from api.db.db_models import UserTenant
 from api.db.services.user_service import UserTenantService, UserService
@@ -22,7 +21,7 @@ from api.db.services.user_service import UserTenantService, UserService
 from common.constants import RetCode, StatusEnum
 from common.misc_utils import get_uuid
 from common.time_utils import delta_seconds
-from api.utils.api_utils import get_json_result, validate_request, server_error_response, get_data_error_result
+from api.utils.api_utils import get_data_error_result, get_json_result, request_json, server_error_response, validate_request
 from api.utils.web_utils import send_invite_email
 from common import settings
 from api.apps import smtp_mail_server, login_required, current_user
@@ -56,7 +55,7 @@ async def create(tenant_id):
             message='No authorization.',
             code=RetCode.AUTHENTICATION_ERROR)
 
-    req = await request.json
+    req = await request_json()
     invite_user_email = req["email"]
     invite_users = UserService.query(email=invite_user_email)
     if not invite_users:

api/apps/user_app.py

@@ -39,6 +39,7 @@ from common.connection_utils import construct_response
 from api.utils.api_utils import (
     get_data_error_result,
     get_json_result,
+    request_json,
     server_error_response,
     validate_request,
 )
@@ -57,6 +58,7 @@ from api.utils.web_utils import (
     captcha_key,
 )
 from common import settings
+from common.http_client import async_request
 
 
 @manager.route("/login", methods=["POST", "GET"])  # noqa: F821
@@ -90,7 +92,7 @@ async def login():
         schema:
           type: object
     """
-    json_body = await request.json
+    json_body = await request_json()
     if not json_body:
         return get_json_result(data=False, code=RetCode.AUTHENTICATION_ERROR, message="Unauthorized!")
 
@@ -136,7 +138,7 @@ async def login():
 
 
 @manager.route("/login/channels", methods=["GET"])  # noqa: F821
-def get_login_channels():
+async def get_login_channels():
     """
     Get all supported authentication channels.
     """
@@ -157,7 +159,7 @@ def get_login_channels():
 
 
 @manager.route("/login/<channel>", methods=["GET"])  # noqa: F821
-def oauth_login(channel):
+async def oauth_login(channel):
     channel_config = settings.OAUTH_CONFIG.get(channel)
     if not channel_config:
         raise ValueError(f"Invalid channel name: {channel}")
@@ -170,7 +172,7 @@ def oauth_login(channel):
 
 
 @manager.route("/oauth/callback/<channel>", methods=["GET"])  # noqa: F821
-def oauth_callback(channel):
+async def oauth_callback(channel):
     """
     Handle the OAuth/OIDC callback for various channels dynamically.
     """
@@ -192,7 +194,10 @@ def oauth_callback(channel):
             return redirect("/?error=missing_code")
 
         # Exchange authorization code for access token
-        token_info = auth_cli.exchange_code_for_token(code)
+        if hasattr(auth_cli, "async_exchange_code_for_token"):
+            token_info = await auth_cli.async_exchange_code_for_token(code)
+        else:
+            token_info = auth_cli.exchange_code_for_token(code)
         access_token = token_info.get("access_token")
         if not access_token:
             return redirect("/?error=token_failed")
@@ -200,7 +205,10 @@ def oauth_callback(channel):
         id_token = token_info.get("id_token")
 
         # Fetch user info
-        user_info = auth_cli.fetch_user_info(access_token, id_token=id_token)
+        if hasattr(auth_cli, "async_fetch_user_info"):
+            user_info = await auth_cli.async_fetch_user_info(access_token, id_token=id_token)
+        else:
+            user_info = auth_cli.fetch_user_info(access_token, id_token=id_token)
         if not user_info.email:
             return redirect("/?error=email_missing")
 
@@ -259,7 +267,7 @@ def oauth_callback(channel):
 
 
 @manager.route("/github_callback", methods=["GET"])  # noqa: F821
-def github_callback():
+async def github_callback():
     """
     **Deprecated**, Use `/oauth/callback/<channel>` instead.
 
@@ -279,9 +287,8 @@ def github_callback():
         schema:
           type: object
     """
-    import requests
+    res = await async_request(
-
+        "POST",
-    res = requests.post(
         settings.GITHUB_OAUTH.get("url"),
         data={
             "client_id": settings.GITHUB_OAUTH.get("client_id"),
@@ -299,7 +306,7 @@ def github_callback():
 
     session["access_token"] = res["access_token"]
     session["access_token_from"] = "github"
-    user_info = user_info_from_github(session["access_token"])
+    user_info = await user_info_from_github(session["access_token"])
     email_address = user_info["email"]
     users = UserService.query(email=email_address)
     user_id = get_uuid()
@@ -348,7 +355,7 @@ def github_callback():
 
 
 @manager.route("/feishu_callback", methods=["GET"])  # noqa: F821
-def feishu_callback():
+async def feishu_callback():
     """
     Feishu OAuth callback endpoint.
     ---
@@ -366,9 +373,8 @@ def feishu_callback():
         schema:
           type: object
     """
-    import requests
+    app_access_token_res = await async_request(
-
+        "POST",
-    app_access_token_res = requests.post(
         settings.FEISHU_OAUTH.get("app_access_token_url"),
         data=json.dumps(
             {
@@ -382,7 +388,8 @@ def feishu_callback():
     if app_access_token_res["code"] != 0:
         return redirect("/?error=%s" % app_access_token_res)
 
-    res = requests.post(
+    res = await async_request(
+        "POST",
         settings.FEISHU_OAUTH.get("user_access_token_url"),
         data=json.dumps(
             {
@@ -403,7 +410,7 @@ def feishu_callback():
         return redirect("/?error=contact:user.email:readonly not in scope")
     session["access_token"] = res["data"]["access_token"]
     session["access_token_from"] = "feishu"
-    user_info = user_info_from_feishu(session["access_token"])
+    user_info = await user_info_from_feishu(session["access_token"])
     email_address = user_info["email"]
     users = UserService.query(email=email_address)
     user_id = get_uuid()
@@ -451,36 +458,34 @@ def feishu_callback():
     return redirect("/?auth=%s" % user.get_id())
 
 
-def user_info_from_feishu(access_token):
+async def user_info_from_feishu(access_token):
-    import requests
-
     headers = {
         "Content-Type": "application/json; charset=utf-8",
         "Authorization": f"Bearer {access_token}",
     }
-    res = requests.get("https://open.feishu.cn/open-apis/authen/v1/user_info", headers=headers)
+    res = await async_request("GET", "https://open.feishu.cn/open-apis/authen/v1/user_info", headers=headers)
     user_info = res.json()["data"]
     user_info["email"] = None if user_info.get("email") == "" else user_info["email"]
     return user_info
 
 
-def user_info_from_github(access_token):
+async def user_info_from_github(access_token):
-    import requests
-
     headers = {"Accept": "application/json", "Authorization": f"token {access_token}"}
-    res = requests.get(f"https://api.github.com/user?access_token={access_token}", headers=headers)
+    res = await async_request("GET", f"https://api.github.com/user?access_token={access_token}", headers=headers)
     user_info = res.json()
-    email_info = requests.get(
+    email_info_response = await async_request(
+        "GET",
         f"https://api.github.com/user/emails?access_token={access_token}",
         headers=headers,
-    ).json()
+    )
+    email_info = email_info_response.json()
     user_info["email"] = next((email for email in email_info if email["primary"]), None)["email"]
     return user_info
 
 
 @manager.route("/logout", methods=["GET"])  # noqa: F821
 @login_required
-def log_out():
+async def log_out():
     """
     User logout endpoint.
     ---
@@ -531,7 +536,7 @@ async def setting_user():
           type: object
     """
     update_dict = {}
-    request_data = await request.json
+    request_data = await request_json()
     if request_data.get("password"):
         new_password = request_data.get("new_password")
         if not check_password_hash(current_user.password, decrypt(request_data["password"])):
@@ -570,7 +575,7 @@ async def setting_user():
 
 @manager.route("/info", methods=["GET"])  # noqa: F821
 @login_required
-def user_profile():
+async def user_profile():
     """
     Get user profile information.
     ---
@@ -698,7 +703,7 @@ async def user_add():
             code=RetCode.OPERATING_ERROR,
         )
 
-    req = await request.json
+    req = await request_json()
     email_address = req["email"]
 
     # Validate the email address
@@ -755,7 +760,7 @@ async def user_add():
 
 @manager.route("/tenant_info", methods=["GET"])  # noqa: F821
 @login_required
-def tenant_info():
+async def tenant_info():
     """
     Get tenant information.
     ---
@@ -831,14 +836,14 @@ async def set_tenant_info():
         schema:
           type: object
     """
-    req = await request.json
+    req = await request_json()
     try:
         tid = req.pop("tenant_id")
         TenantService.update_by_id(tid, req)
         return get_json_result(data=True)
     except Exception as e:
         return server_error_response(e)
-        
+
 
 @manager.route("/forget/captcha", methods=["GET"])  # noqa: F821
 async def forget_get_captcha():
@@ -875,7 +880,7 @@ async def forget_send_otp():
     - Verify the image captcha stored at captcha:{email} (case-insensitive).
     - On success, generate an email OTP (A–Z with length = OTP_LENGTH), store hash + salt (and timestamp) in Redis with TTL, reset attempts and cooldown, and send the OTP via email.
     """
-    req = await request.get_json()
+    req = await request_json()
     email = req.get("email") or ""
     captcha = (req.get("captcha") or "").strip()
 
@@ -931,7 +936,7 @@ async def forget_send_otp():
             )
         except Exception:
             return get_json_result(data=False, code=RetCode.SERVER_ERROR, message="failed to send email")
-        
+
     return get_json_result(data=True, code=RetCode.SUCCESS, message="verification passed, email sent")
 
 
@@ -941,7 +946,7 @@ async def forget():
     POST: Verify email + OTP and reset password, then log the user in.
     Request JSON: { email, otp, new_password, confirm_new_password }
     """
-    req = await request.get_json()
+    req = await request_json()
     email = req.get("email") or ""
     otp = (req.get("otp") or "").strip()
     new_pwd = req.get("new_password")
@@ -1006,4 +1011,4 @@ async def forget():
     user.update_date = (datetime_format(datetime.now()),)
     user.save()
     msg = "Password reset successful. Logged in."
-    return construct_response(data=user.to_json(), auth=user.get_id(), message=msg)
+    return await construct_response(data=user.to_json(), auth=user.get_id(), message=msg)

api/ragflow_server.py

@@ -25,7 +25,6 @@ import logging
 import os
 import signal
 import sys
-import time
 import traceback
 import threading
 import uuid
@@ -69,7 +68,7 @@ def signal_handler(sig, frame):
     logging.info("Received interrupt signal, shutting down...")
     shutdown_all_mcp_sessions()
     stop_event.set()
-    time.sleep(1)
+    stop_event.wait(1)
     sys.exit(0)
 
 if __name__ == '__main__':
@@ -163,5 +162,5 @@ if __name__ == '__main__':
     except Exception:
         traceback.print_exc()
         stop_event.set()
-        time.sleep(1)
+        stop_event.wait(1)
         os.kill(os.getpid(), signal.SIGKILL)

api/utils/api_utils.py

@@ -44,12 +44,22 @@ from common import settings
 
 requests.models.complexjson.dumps = functools.partial(json.dumps, cls=CustomJSONEncoder)
 
+async def _coerce_request_data() -> dict:
+    """Fetch JSON body with sane defaults; fallback to form data."""
+    try:
+        payload = await request.get_json(force=True, silent=True)
+    except Exception:
+        payload = None
+    if payload is None:
+        try:
+            payload = (await request.form).to_dict()
+        except Exception:
+            payload = {}
+    return payload or {}
+
 
 async def request_json():
-    try:
+    return await _coerce_request_data()
-        return await request.json
-    except Exception:
-        return {}
 
 def serialize_for_json(obj):
     """
@@ -137,7 +147,7 @@ def validate_request(*args, **kwargs):
     def wrapper(func):
         @wraps(func)
         async def decorated_function(*_args, **_kwargs):
-            errs = process_args(await request.json or (await request.form).to_dict())
+            errs = process_args(await _coerce_request_data())
             if errs:
                 return get_json_result(code=RetCode.ARGUMENT_ERROR, message=errs)
             if inspect.iscoroutinefunction(func):
@@ -152,7 +162,7 @@ def validate_request(*args, **kwargs):
 def not_allowed_parameters(*params):
     def decorator(func):
         async def wrapper(*args, **kwargs):
-            input_arguments = await request.json or (await request.form).to_dict()
+            input_arguments = await _coerce_request_data()
             for param in params:
                 if param in input_arguments:
                     return get_json_result(code=RetCode.ARGUMENT_ERROR, message=f"Parameter {param} isn't allowed")

common/http_client.py

@@ -0,0 +1,157 @@
+# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import asyncio
+import logging
+import os
+import time
+from typing import Any, Dict, Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Default knobs; keep conservative to avoid unexpected behavioural changes.
+DEFAULT_TIMEOUT = float(os.environ.get("HTTP_CLIENT_TIMEOUT", "15"))
+# Align with requests default: follow redirects with a max of 30 unless overridden.
+DEFAULT_FOLLOW_REDIRECTS = bool(int(os.environ.get("HTTP_CLIENT_FOLLOW_REDIRECTS", "1")))
+DEFAULT_MAX_REDIRECTS = int(os.environ.get("HTTP_CLIENT_MAX_REDIRECTS", "30"))
+DEFAULT_MAX_RETRIES = int(os.environ.get("HTTP_CLIENT_MAX_RETRIES", "2"))
+DEFAULT_BACKOFF_FACTOR = float(os.environ.get("HTTP_CLIENT_BACKOFF_FACTOR", "0.5"))
+DEFAULT_PROXY = os.environ.get("HTTP_CLIENT_PROXY")
+DEFAULT_USER_AGENT = os.environ.get("HTTP_CLIENT_USER_AGENT", "ragflow-http-client")
+
+
+def _clean_headers(headers: Optional[Dict[str, str]], auth_token: Optional[str] = None) -> Optional[Dict[str, str]]:
+    merged_headers: Dict[str, str] = {}
+    if DEFAULT_USER_AGENT:
+        merged_headers["User-Agent"] = DEFAULT_USER_AGENT
+    if auth_token:
+        merged_headers["Authorization"] = auth_token
+    if headers is None:
+        return merged_headers or None
+    merged_headers.update({str(k): str(v) for k, v in headers.items() if v is not None})
+    return merged_headers or None
+
+
+def _get_delay(backoff_factor: float, attempt: int) -> float:
+    return backoff_factor * (2**attempt)
+
+
+async def async_request(
+    method: str,
+    url: str,
+    *,
+    timeout: float | httpx.Timeout | None = None,
+    follow_redirects: bool | None = None,
+    max_redirects: Optional[int] = None,
+    headers: Optional[Dict[str, str]] = None,
+    auth_token: Optional[str] = None,
+    retries: Optional[int] = None,
+    backoff_factor: Optional[float] = None,
+    proxies: Any = None,
+    **kwargs: Any,
+) -> httpx.Response:
+    """Lightweight async HTTP wrapper using httpx.AsyncClient with safe defaults."""
+    timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
+    follow_redirects = DEFAULT_FOLLOW_REDIRECTS if follow_redirects is None else follow_redirects
+    max_redirects = DEFAULT_MAX_REDIRECTS if max_redirects is None else max_redirects
+    retries = DEFAULT_MAX_RETRIES if retries is None else max(retries, 0)
+    backoff_factor = DEFAULT_BACKOFF_FACTOR if backoff_factor is None else backoff_factor
+    headers = _clean_headers(headers, auth_token=auth_token)
+    proxies = DEFAULT_PROXY if proxies is None else proxies
+
+    async with httpx.AsyncClient(
+        timeout=timeout,
+        follow_redirects=follow_redirects,
+        max_redirects=max_redirects,
+        proxies=proxies,
+    ) as client:
+        last_exc: Exception | None = None
+        for attempt in range(retries + 1):
+            try:
+                start = time.monotonic()
+                response = await client.request(method=method, url=url, headers=headers, **kwargs)
+                duration = time.monotonic() - start
+                logger.debug(f"async_request {method} {url} -> {response.status_code} in {duration:.3f}s")
+                return response
+            except httpx.RequestError as exc:
+                last_exc = exc
+                if attempt >= retries:
+                    logger.warning(f"async_request exhausted retries for {method} {url}: {exc}")
+                    raise
+                delay = _get_delay(backoff_factor, attempt)
+                logger.warning(f"async_request attempt {attempt + 1}/{retries + 1} failed for {method} {url}: {exc}; retrying in {delay:.2f}s")
+                await asyncio.sleep(delay)
+        raise last_exc  # pragma: no cover
+
+
+def sync_request(
+    method: str,
+    url: str,
+    *,
+    timeout: float | httpx.Timeout | None = None,
+    follow_redirects: bool | None = None,
+    max_redirects: Optional[int] = None,
+    headers: Optional[Dict[str, str]] = None,
+    auth_token: Optional[str] = None,
+    retries: Optional[int] = None,
+    backoff_factor: Optional[float] = None,
+    proxies: Any = None,
+    **kwargs: Any,
+) -> httpx.Response:
+    """Synchronous counterpart to async_request, for CLI/tests or sync contexts."""
+    timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
+    follow_redirects = DEFAULT_FOLLOW_REDIRECTS if follow_redirects is None else follow_redirects
+    max_redirects = DEFAULT_MAX_REDIRECTS if max_redirects is None else max_redirects
+    retries = DEFAULT_MAX_RETRIES if retries is None else max(retries, 0)
+    backoff_factor = DEFAULT_BACKOFF_FACTOR if backoff_factor is None else backoff_factor
+    headers = _clean_headers(headers, auth_token=auth_token)
+    proxies = DEFAULT_PROXY if proxies is None else proxies
+
+    with httpx.Client(
+        timeout=timeout,
+        follow_redirects=follow_redirects,
+        max_redirects=max_redirects,
+        proxies=proxies,
+    ) as client:
+        last_exc: Exception | None = None
+        for attempt in range(retries + 1):
+            try:
+                start = time.monotonic()
+                response = client.request(method=method, url=url, headers=headers, **kwargs)
+                duration = time.monotonic() - start
+                logger.debug(f"sync_request {method} {url} -> {response.status_code} in {duration:.3f}s")
+                return response
+            except httpx.RequestError as exc:
+                last_exc = exc
+                if attempt >= retries:
+                    logger.warning(f"sync_request exhausted retries for {method} {url}: {exc}")
+                    raise
+                delay = _get_delay(backoff_factor, attempt)
+                logger.warning(f"sync_request attempt {attempt + 1}/{retries + 1} failed for {method} {url}: {exc}; retrying in {delay:.2f}s")
+                time.sleep(delay)
+        raise last_exc  # pragma: no cover
+
+
+__all__ = [
+    "async_request",
+    "sync_request",
+    "DEFAULT_TIMEOUT",
+    "DEFAULT_FOLLOW_REDIRECTS",
+    "DEFAULT_MAX_REDIRECTS",
+    "DEFAULT_MAX_RETRIES",
+    "DEFAULT_BACKOFF_FACTOR",
+    "DEFAULT_PROXY",
+    "DEFAULT_USER_AGENT",
+]