diff --git a/dashscope/api_entities/aio_session.py b/dashscope/api_entities/aio_session.py
index 0bd702a..3e7a47e 100644
--- a/dashscope/api_entities/aio_session.py
+++ b/dashscope/api_entities/aio_session.py
@@ -7,6 +7,7 @@
 is created once and shared across all sessions.
 """
 import asyncio
+import atexit
 import ssl
 import threading
 import weakref
@@ -47,45 +48,55 @@ async def get_shared_aio_session() -> aiohttp.ClientSession:
         connector = aiohttp.TCPConnector(ssl=get_ssl_context())
         session = aiohttp.ClientSession(connector=connector, trust_env=True)
 
-        # Register finalizer to close session when it's garbage collected.
-        # This prevents resource leaks even if
-        # close_shared_aio_session() is not called.
-        weakref.finalize(session, _sync_close_session, id(session))
-
         _aio_sessions[loop] = session
+        # Register GC-safe finalizer to close session when loop is collected
+        weakref.finalize(session, _sync_close_session, id(session))
     return session
 
 
 def _sync_close_session(session_id: int) -> None:
-    """Synchronous callback when session is garbage collected.
-
-    Note: This runs during GC, so we cannot await. We try to close
-    synchronously if possible, otherwise log a warning.
-    """
+    """GC callback to safely close a session outside async context."""
     try:
         loop = asyncio.get_event_loop()
-        if not loop.is_closed() and not loop.is_running():
-            # Loop exists and is not running, we can use it
-            # Find the session in our cache
-            with _lock:
-                for stored_session in _aio_sessions.values():
-                    if (
-                        id(stored_session) == session_id
-                        and not stored_session.closed
-                    ):
-                        try:
-                            loop.run_until_complete(stored_session.close())
-                        except Exception:
-                            pass
-                        break
+        if loop.is_closed():
+            return
+        if loop.is_running():
+            asyncio.ensure_future(_async_close_by_id(session_id))
+        else:
+            loop.run_until_complete(_async_close_by_id(session_id))
     except RuntimeError:
-        # No event loop available, can't close asynchronously
-        pass
-    except Exception:
-        # Ignore errors during cleanup
         pass
 
 
+async def _async_close_by_id(session_id: int) -> None:
+    """Close session by ID, safe against already-collected sessions."""
+    with _lock:
+        for loop, session in list(_aio_sessions.items()):
+            if id(session) == session_id and not session.closed:
+                await session.close()
+                _aio_sessions.pop(loop, None)
+                return
+
+
+def _atexit_cleanup() -> None:
+    """Cleanup all sessions at interpreter exit."""
+    with _lock:
+        sessions = list(_aio_sessions.items())
+        _aio_sessions.clear()
+
+    for loop, session in sessions:
+        if not session.closed:
+            try:
+                if not loop.is_closed() and not loop.is_running():
+                    loop.run_until_complete(session.close())
+            except Exception:
+                pass
+
+
+# Register atexit handler
+atexit.register(_atexit_cleanup)
+
+
 async def close_shared_aio_session() -> None:
     """Close the shared session for the current event loop."""
     loop = asyncio.get_running_loop()
diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py
index d220ffb..a3c8752 100644
--- a/dashscope/api_entities/aiohttp_request.py
+++ b/dashscope/api_entities/aiohttp_request.py
@@ -78,7 +78,7 @@ def __init__(
             self.headers["X-Accel-Buffering"] = "no"
             self.headers["X-DashScope-SSE"] = "enable"
         if self.query:
-            self.url = self.url.replace("api", "api-task")
+            self.url = self.url.replace("/api/", "/api-task/", 1)
             self.url += f"{task_id}"
         if timeout is None:
             self.timeout = DEFAULT_REQUEST_TIMEOUT_SECONDS
@@ -89,7 +89,7 @@ def add_header(self, key, value):
         self.headers[key] = value
 
     def add_headers(self, headers):
-        self.headers = {**self.headers, **headers}
+        self.headers.update(headers)
 
     def call(self):
         response = async_to_sync(self._handle_request())
@@ -142,6 +142,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
         response: aiohttp.ClientResponse,
     ):
         request_id = ""
+        headers = dict(response.headers)
         if (
             response.status == HTTPStatus.OK
             and self.stream
@@ -167,6 +168,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                         status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                         code="Unknown",
                         message=data,
+                        headers=headers,
                     )
                     continue
                 if is_error:
@@ -175,6 +177,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                         status_code=status_code,
                         code=msg["code"],
                         message=msg["message"],
+                        headers=headers,
                     )
                 else:
                     yield DashScopeAPIResponse(
@@ -182,6 +185,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                         status_code=HTTPStatus.OK,
                         output=output,
                         usage=usage,
+                        headers=headers,
                     )
         elif (
             response.status == HTTPStatus.OK
@@ -201,6 +205,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                 request_id=request_id,
                 status_code=HTTPStatus.OK,
                 output=output,
+                headers=headers,
             )
         elif response.status == HTTPStatus.OK:
             json_content = await response.json()
@@ -217,6 +222,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                 status_code=HTTPStatus.OK,
                 output=output,
                 usage=usage,
+                headers=headers,
             )
         else:
             if "application/json" in response.content_type:
@@ -243,6 +249,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                     status_code=response.status,
                     code=error["code"],
                     message=message,
+                    headers=headers,
                 )
             else:
                 msg = await response.read()
@@ -251,6 +258,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                     status_code=response.status,
                     code="Unknown",
                     message=msg.decode("utf-8"),
+                    headers=headers,
                 )
 
     # pylint: disable=too-many-branches
@@ -258,10 +266,8 @@ async def _handle_request(self):
         try:
             if self._external_aio_session is not None:
                 session = self._external_aio_session
-                should_close = False
             else:
                 session = await get_shared_aio_session()
-                should_close = False
 
             if self.stream:
                 request_timeout = aiohttp.ClientTimeout(
@@ -271,52 +277,48 @@ async def _handle_request(self):
             else:
                 request_timeout = aiohttp.ClientTimeout(total=self.timeout)
 
-            try:
-                logger.debug("Starting request: %s", self.url)
-                if self.method == HTTPMethod.POST:
-                    is_form, obj = False, {}
-                    if hasattr(self, "data") and self.data is not None:
-                        is_form, obj = self.data.get_aiohttp_payload()
-                    if is_form:
-                        headers = {**self.headers, **obj.headers}
-                        response = await session.post(
-                            url=self.url,
-                            data=obj,
-                            headers=headers,
-                            timeout=request_timeout,
-                        )
-                    else:
-                        body = json.dumps(obj, ensure_ascii=False).encode(
-                            "utf-8",
-                        )
-                        response = await session.request(
-                            "POST",
-                            url=self.url,
-                            data=body,
-                            headers=self.headers,
-                            timeout=request_timeout,
-                        )
-                elif self.method == HTTPMethod.GET:
-                    params = {}
-                    if hasattr(self, "data") and self.data is not None:
-                        params = getattr(self.data, "parameters", {})
-                    response = await session.get(
+            logger.debug("Starting request: %s", self.url)
+            if self.method == HTTPMethod.POST:
+                is_form, obj = False, {}
+                if hasattr(self, "data") and self.data is not None:
+                    is_form, obj = self.data.get_aiohttp_payload()
+                if is_form:
+                    headers = {**self.headers, **obj.headers}
+                    response = await session.post(
                         url=self.url,
-                        params=params,
-                        headers=self.headers,
+                        data=obj,
+                        headers=headers,
                         timeout=request_timeout,
                     )
                 else:
-                    raise UnsupportedHTTPMethod(
-                        f"Unsupported http method: {self.method}",
+                    body = json.dumps(obj, ensure_ascii=False).encode(
+                        "utf-8",
                     )
-                logger.debug("Response returned: %s", self.url)
-                async with response:
-                    async for rsp in self._handle_response(response):
-                        yield rsp
-            finally:
-                if should_close:
-                    await session.close()
-        except Exception as e:
-            logger.debug(e)
-            raise e
+                    response = await session.request(
+                        "POST",
+                        url=self.url,
+                        data=body,
+                        headers=self.headers,
+                        timeout=request_timeout,
+                    )
+            elif self.method == HTTPMethod.GET:
+                params = {}
+                if hasattr(self, "data") and self.data is not None:
+                    params = getattr(self.data, "parameters", {})
+                response = await session.get(
+                    url=self.url,
+                    params=params,
+                    headers=self.headers,
+                    timeout=request_timeout,
+                )
+            else:
+                raise UnsupportedHTTPMethod(
+                    f"Unsupported http method: {self.method}",
+                )
+            logger.debug("Response returned: %s", self.url)
+            async with response:
+                async for rsp in self._handle_response(response):
+                    yield rsp
+        except Exception:
+            logger.exception("Request failed")
+            raise
diff --git a/dashscope/api_entities/api_request_data.py b/dashscope/api_entities/api_request_data.py
index 50ab01c..c6ea54a 100644
--- a/dashscope/api_entities/api_request_data.py
+++ b/dashscope/api_entities/api_request_data.py
@@ -2,6 +2,7 @@
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
 import json
+from typing import Optional
 from urllib.parse import urlencode
 
 import aiohttp
@@ -50,10 +51,10 @@ def add_parameters(self, **params):
     def add_resources(self, resources):
         self.resources = resources
 
-    def to_request_object(self) -> str:
-        """Convert data to json, called from http request.
+    def to_request_object(self) -> dict:
+        """Convert data to request dict, called from http request.
         Returns:
-            str: Json string.
+            dict: Request payload.
         """
         self.input = next(self._input_resolver)
         o = {
@@ -61,8 +62,7 @@ def to_request_object(self) -> str:
             for k, v in self.__dict__.items()
             if not (
                 k.startswith("_")
-                or k.startswith("task")
-                or k.startswith("function")
+                or k in ("task_group", "task", "function")
                 or v is None
             )
         }
@@ -92,15 +92,6 @@ def get_aiohttp_payload(self):
                 json.dumps(data["parameters"], ensure_ascii=False),
             )
             return True, form()
-            # pylint: disable=unreachable,pointless-string-statement
-            """
-            mp_writer = aiohttp.MultipartWriter('mixed')
-            mp_writer.append('model=%s'%self.model)
-            mp_writer.append('input=%s' % json.dumps(self._input))
-            mp_writer.append('parameters=%s'%json.dumps(self.parameters))
-            mp_writer.append(form())
-            return True, mp_writer
-            """
         else:
             return False, data
 
@@ -143,15 +134,7 @@ def get_websocket_continue_data(self):
         for content in self._input_resolver:
             yield content
 
-    def _to_json_only_data(self) -> str:
-        o = {
-            k: v
-            for k, v in self.__dict__.items()
-            if not (k.startswith("_") or k.startswith("param"))
-        }
-        return json.dumps(o, default=lambda o: o.__dict__)
-
-    def get_batch_binary_data(self) -> bytes:  # type: ignore[return]
+    def get_batch_binary_data(self) -> Optional[bytes]:
         """Get binary data. used in streaming mode none and
            out (input is not streaming), we send data in one package.
            In this case only has one field input.
@@ -161,12 +144,14 @@ def get_batch_binary_data(self) -> bytes:  # type: ignore[return]
         """
         for content in self._input_resolver:
             return content
+        return None
 
-    def _only_parameters(self) -> str:
+    def _only_parameters(self) -> dict:
         temp_input = None
-        if "raw_input" in self.parameters:
-            temp_input = self.parameters.pop("raw_input")
-        obj = {"model": self.model, "parameters": self.parameters, "input": {}}
+        params = dict(self.parameters)
+        if "raw_input" in params:
+            temp_input = params.pop("raw_input")
+        obj = {"model": self.model, "parameters": params, "input": {}}
         if temp_input is not None:
             obj["input"] = temp_input
         if self.task is not None:
diff --git a/dashscope/api_entities/dashscope_response.py b/dashscope/api_entities/dashscope_response.py
index 58ade6b..0060dcf 100644
--- a/dashscope/api_entities/dashscope_response.py
+++ b/dashscope/api_entities/dashscope_response.py
@@ -6,6 +6,8 @@
 from http import HTTPStatus
 from typing import Any, Dict, List, Union
 
+_MISSING = object()
+
 
 @dataclass(init=False)
 class DictMixin(dict):
@@ -14,9 +16,6 @@ class DictMixin(dict):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def __getitem__(self, key):
-        return super().__getitem__(key)
-
     def __copy__(self):
         return type(self)(**self)
 
@@ -28,27 +27,14 @@ def __deepcopy__(self, memo):
             memo[id_self] = _copy
         return _copy
 
-    def __setitem__(self, key, value):
-        return super().__setitem__(key, value)
-
-    def __delitem__(self, key):
-        return super().__delitem__(key)
-
-    def get(self, key, default=None):
-        return super().get(key, default)
-
-    def setdefault(self, key, default=None):
-        return super().setdefault(key, default)
-
-    def pop(self, key, default: Any):  # type: ignore[override]
+    def pop(self, key, default: Any = _MISSING):
+        if default is _MISSING:
+            return super().pop(key)
         return super().pop(key, default)
 
     def update(self, **kwargs):
         super().update(**kwargs)
 
-    def __contains__(self, key):
-        return super().__contains__(key)
-
     def copy(self):
         return type(self)(self)
 
@@ -164,7 +150,10 @@ def from_generation_response(cls, response: DictMixin):
             content = response.output["text"]
             return Message(role=Role.ASSISTANT, content=content)
         else:
-            return response.output.choices[0]["message"]
+            choices = response.output.get("choices", [])
+            if not choices:
+                return None
+            return choices[0].get("message")
 
     @classmethod
     def from_conversation_response(cls, response: DictMixin):
diff --git a/dashscope/api_entities/encryption.py b/dashscope/api_entities/encryption.py
index 59698e4..e1017c0 100644
--- a/dashscope/api_entities/encryption.py
+++ b/dashscope/api_entities/encryption.py
@@ -15,6 +15,7 @@
 from dashscope.common.constants import (
     ENCRYPTION_AES_SECRET_KEY_BYTES,
     ENCRYPTION_AES_IV_LENGTH,
+    DEFAULT_REQUEST_TIMEOUT_SECONDS,
 )
 from dashscope.common.logging import logger
 
@@ -91,7 +92,11 @@ def _get_public_keys():
             "Authorization": f"Bearer {dashscope.api_key}",
         }
 
-        response = requests.get(url, headers=headers)
+        response = requests.get(
+            url,
+            headers=headers,
+            timeout=DEFAULT_REQUEST_TIMEOUT_SECONDS,
+        )
         if response.status_code != 200:
             logger.error("exceptional public key response: %s", response)
             return None
diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py
index 285c6ab..cdfd18f 100644
--- a/dashscope/api_entities/http_request.py
+++ b/dashscope/api_entities/http_request.py
@@ -2,6 +2,7 @@
 # Copyright (c) Alibaba, Inc. and its affiliates.
 import datetime
 import json
+import threading
 from http import HTTPStatus
 from typing import Optional, Dict, Union
 
@@ -26,6 +27,24 @@
 )
 from dashscope.api_entities.encryption import Encryption
 
+# Shared synchronous session for connection pooling
+_shared_sync_session: Optional[requests.Session] = None
+_shared_sync_session_lock = threading.Lock()
+
+
+def _get_shared_sync_session() -> requests.Session:
+    """Return a shared requests.Session for connection pooling.
+
+    The session is created lazily and reused for the process lifetime;
+    it is never closed explicitly.
+    """
+    global _shared_sync_session
+    if _shared_sync_session is None:
+        with _shared_sync_session_lock:
+            if _shared_sync_session is None:
+                _shared_sync_session = requests.Session()
+    return _shared_sync_session
+
 
 class HttpRequest(AioBaseRequest):
     def __init__(
@@ -123,7 +142,7 @@ def __init__(
             self.headers["X-Accel-Buffering"] = "no"
             self.headers["X-DashScope-SSE"] = "enable"
         if self.query:
-            self.url = self.url.replace("api", "api-task")
+            self.url = self.url.replace("/api/", "/api-task/", 1)
             self.url += f"{task_id}"
         if timeout is None:
             self.timeout = DEFAULT_REQUEST_TIMEOUT_SECONDS
@@ -134,7 +153,7 @@ def add_header(self, key, value):
         self.headers[key] = value
 
     def add_headers(self, headers):
-        self.headers = {**self.headers, **headers}
+        self.headers.update(headers)
 
     def call(self):
         response = self._handle_request()
@@ -227,9 +246,9 @@ async def _handle_aio_request(self):  # pylint: disable=too-many-branches
             finally:
                 if should_close:
                     await session.close()
-        except Exception as e:
-            logger.debug(e)
-            raise e
+        except Exception:
+            logger.exception("Request failed")
+            raise
 
     @staticmethod
     def __handle_parameters(params: dict) -> dict:
@@ -435,10 +454,10 @@ def _handle_response(  # pylint: disable=too-many-branches
             logger.debug("Response: %s", json_content)
             output = None
             usage = None
-            if "task_id" in json_content:
-                output = {"task_id": json_content["task_id"]}
             if "output" in json_content:
                 output = json_content["output"]
+            elif "task_id" in json_content:
+                output = {"task_id": json_content["task_id"]}
             if "usage" in json_content:
                 usage = json_content["usage"]
             if "request_id" in json_content:
@@ -461,61 +480,54 @@ def _handle_response(  # pylint: disable=too-many-branches
     def _handle_request(self):  # pylint: disable=too-many-branches
         try:
             # Use external session if provided,
-            # otherwise create temporary session
+            # otherwise use shared session with connection pooling
             if self._external_session is not None:
                 session = self._external_session
-                should_close = False
             else:
-                session = requests.Session()
-                should_close = True
+                session = _get_shared_sync_session()
 
-            try:
-                if self.method == HTTPMethod.POST:
-                    is_form, form, obj = False, None, {}
-                    if hasattr(self, "data") and self.data is not None:
-                        is_form, form, obj = self.data.get_http_payload()
-                    if is_form:
-                        headers = {**self.headers}
-                        headers.pop("Content-Type")
-                        response = session.post(
-                            url=self.url,
-                            data=obj,
-                            files=form,
-                            headers=headers,
-                            timeout=self.timeout,
-                        )
-                    else:
-                        logger.debug("Request body: %s", obj)
-                        body = json.dumps(obj, ensure_ascii=False).encode(
-                            "utf-8",
-                        )
-                        response = session.post(
-                            url=self.url,
-                            stream=self.stream,
-                            data=body,
-                            headers={**self.headers},
-                            timeout=self.timeout,
-                        )
-                elif self.method == HTTPMethod.GET:
-                    params = {}
-                    if hasattr(self, "data") and self.data is not None:
-                        params = getattr(self.data, "parameters", {})
-                    response = session.get(
+            if self.method == HTTPMethod.POST:
+                is_form, form, obj = False, None, {}
+                if hasattr(self, "data") and self.data is not None:
+                    is_form, form, obj = self.data.get_http_payload()
+                if is_form:
+                    headers = {**self.headers}
+                    headers.pop("Content-Type")
+                    response = session.post(
                         url=self.url,
-                        params=params,
-                        headers=self.headers,
+                        data=obj,
+                        files=form,
+                        headers=headers,
                         timeout=self.timeout,
                     )
                 else:
-                    raise UnsupportedHTTPMethod(
-                        f"Unsupported http method: {self.method}",
+                    logger.debug("Request body: %s", obj)
+                    body = json.dumps(obj, ensure_ascii=False).encode(
+                        "utf-8",
                     )
-                for rsp in self._handle_response(response):
-                    yield rsp
-            finally:
-                # Only close if we created the session
-                if should_close:
-                    session.close()
-        except Exception as e:
-            logger.debug(e)
-            raise e
+                    response = session.post(
+                        url=self.url,
+                        stream=self.stream,
+                        data=body,
+                        headers={**self.headers},
+                        timeout=self.timeout,
+                    )
+            elif self.method == HTTPMethod.GET:
+                params = {}
+                if hasattr(self, "data") and self.data is not None:
+                    params = getattr(self.data, "parameters", {})
+                response = session.get(
+                    url=self.url,
+                    params=params,
+                    headers=self.headers,
+                    timeout=self.timeout,
+                )
+            else:
+                raise UnsupportedHTTPMethod(
+                    f"Unsupported http method: {self.method}",
+                )
+            for rsp in self._handle_response(response):
+                yield rsp
+        except Exception:
+            logger.exception("Request failed")
+            raise
diff --git a/dashscope/api_entities/websocket_request.py b/dashscope/api_entities/websocket_request.py
index a3ceb97..13fc853 100644
--- a/dashscope/api_entities/websocket_request.py
+++ b/dashscope/api_entities/websocket_request.py
@@ -72,7 +72,7 @@ def __init__(
         self.is_binary_input = is_binary_input
 
         self.headers = {
-            "Authorization": f"bearer {api_key}",
+            "Authorization": f"Bearer {api_key}",
             **self.headers,  # type: ignore[has-type]
         }
 
@@ -80,9 +80,10 @@ def __init__(
             "streaming": self.ws_stream_mode,
         }
         self.pre_task_id = pre_task_id
+        self.ws = None
 
     def add_headers(self, headers):
-        self.headers = {**self.headers, **headers}
+        self.headers.update(headers)
 
     def call(self):
         response = async_to_sync(self.connection_handler())
@@ -97,8 +98,9 @@ def call(self):
             return output
 
     async def close(self):
-        if self.ws is not None and not self.ws.closed:
-            await self.ws.close()
+        ws = getattr(self, "ws", None)
+        if ws is not None and not ws.closed:
+            await ws.close()
 
     async def aio_call(self):
         response = self.connection_handler()
@@ -112,7 +114,9 @@ async def aio_call(self):
                 pass
             return result
 
-    async def connection_handler(self):  # pylint: disable=too-many-branches
+    async def connection_handler(
+        self,
+    ):  # pylint: disable=too-many-branches,too-many-statements
         try:
             task_id = None
             async with aiohttp.ClientSession(
@@ -124,8 +128,9 @@ async def connection_handler(self):  # pylint: disable=too-many-branches
                 async with session.ws_connect(
                     self.url,
                     headers=self.headers,
-                    heartbeat=6000,
+                    heartbeat=30,
                 ) as ws:
+                    self.ws = ws  # Store ws reference for close() method
                     await self._start_task(ws)  # send start task action.
                     task_id = self.task_headers["task_id"]
                     await self._wait_for_task_started(
@@ -176,15 +181,30 @@ async def connection_handler(self):  # pylint: disable=too-many-branches
                                 message,
                             )
                     else:  # duplex mode
-                        asyncio.create_task(self._send_continue_task_data(ws))
-                        async for is_binary, message in self._receive_streaming_data_task(  # noqa E501  # pylint: disable=line-too-long
-                            ws,
-                        ):
-                            yield self._to_DashScopeAPIResponse(
-                                task_id,
-                                is_binary,
-                                message,
+                        bg_task = asyncio.create_task(
+                            self._send_continue_task_data(ws),
+                        )
+                        try:
+                            async for is_binary, message in self._receive_streaming_data_task(  # noqa E501  # pylint: disable=line-too-long
+                                ws,
+                            ):
+                                yield self._to_DashScopeAPIResponse(
+                                    task_id,
+                                    is_binary,
+                                    message,
+                                )
+                            # Normal completion: wait for the send task.
+                            await bg_task
+                        except BaseException:
+                            # Abnormal exit (error or consumer closed the
+                            # stream early): cancel to avoid leaking it.
+                            if not bg_task.done():
+                                bg_task.cancel()
+                            await asyncio.gather(
+                                bg_task,
+                                return_exceptions=True,
                             )
+                            raise
         except RequestFailure as e:
             yield DashScopeAPIResponse(
                 request_id=e.request_id,
@@ -216,7 +236,7 @@ async def connection_handler(self):  # pylint: disable=too-many-branches
                 code=code,
                 message=msg,
             )
-        except BaseException as e:
+        except Exception as e:
             logger.exception(e)
             yield DashScopeAPIResponse(
                 request_id="",
diff --git a/dashscope/threads/messages/messages.py b/dashscope/threads/messages/messages.py
index 2420609..10953af 100644
--- a/dashscope/threads/messages/messages.py
+++ b/dashscope/threads/messages/messages.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
+import warnings
 from typing import Dict, List, Optional
 
 from dashscope.client.base_api import (
@@ -14,8 +15,24 @@
 
 __all__ = ["Messages"]
 
+# Deprecation warning message
+_DEPRECATION_MSG = (
+    "The Assistants API (dashscope.threads) is deprecated and will be "
+    "removed in a future release. Please migrate to the Responses API. "
+    "See https://help.aliyun.com/zh/model-studio/"
+    "synchronous-call-api-reference for migration details."
+)
+
 
 class Messages(CreateMixin, ListObjectMixin, GetStatusMixin, UpdateMixin):
+    """
+    .. deprecated::
+        The Messages API (Assistants API) is deprecated and will be removed
+        in a future release. Please migrate to the Responses API.
+        See https://help.aliyun.com/zh/model-studio/
+        synchronous-call-api-reference for migration details.
+    """
+
     SUB_PATH = "messages"  # useless
 
     @classmethod
@@ -90,6 +107,11 @@ def create(
         Returns:
             ThreadMessage: The `ThreadMessage` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         cls.SUB_PATH = f"{thread_id}/messages"
         data = {}
         if not thread_id or not content:
@@ -160,6 +182,11 @@ def get(  # type: ignore[override]
         Returns:
             ThreadMessage: The `ThreadMessage` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not message_id or not thread_id:
             raise InputRequired("thread id, message id are required!")
         response = super().get(
@@ -201,6 +228,11 @@ def list(  # type: ignore[override]
         Returns:
             ThreadMessageList: The `ThreadMessageList` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id:
             raise InputRequired("thread id is required!")
         response = super().list(
diff --git a/dashscope/threads/runs/runs.py b/dashscope/threads/runs/runs.py
index ed5f30f..756d1ed 100644
--- a/dashscope/threads/runs/runs.py
+++ b/dashscope/threads/runs/runs.py
@@ -2,6 +2,7 @@
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
 import time
+import warnings
 from http import HTTPStatus
 from typing import Dict, List, Optional
 
@@ -30,6 +31,14 @@
 
 __all__ = ["Runs"]
 
+# Deprecation warning message
+_DEPRECATION_MSG = (
+    "The Assistants API (dashscope.threads) is deprecated and will be "
+    "removed in a future release. Please migrate to the Responses API. "
+    "See https://help.aliyun.com/zh/model-studio/"
+    "synchronous-call-api-reference for migration details."
+)
+
 
 class Runs(
     CreateMixin,
@@ -38,6 +47,14 @@ class Runs(
     GetStatusMixin,
     UpdateMixin,
 ):
+    """
+    .. deprecated::
+        The Runs API (Assistants API) is deprecated and will be removed
+        in a future release. Please migrate to the Responses API.
+        See https://help.aliyun.com/zh/model-studio/
+        synchronous-call-api-reference for migration details.
+    """
+
     SUB_PATH = "RUNS"  # useless
 
     @classmethod
@@ -57,6 +74,31 @@ def create_thread_and_run(
         api_key: str = None,
         **kwargs,
     ) -> Run:
+        """Create a thread and run it with an assistant.
+
+        Args:
+            assistant_id (str): The assistant id to run.
+            thread (Optional[Dict], optional): The thread to create.
+            model (Optional[str], optional): The model to use.
+            instructions (Optional[str], optional): The instructions.
+            additional_instructions (Optional[str], optional):
+                Additional instructions.
+            tools (Optional[List[Dict]], optional): The tools to use.
+            stream (Optional[bool], optional):
+                Whether to stream. Defaults to False.
+            metadata (Optional[Dict], optional): The metadata.
+            workspace (str, optional): The workspace id.
+            extra_body (Optional[Dict], optional): Extra body parameters.
+            api_key (str, optional): The api key.
+
+        Returns:
+            Run: The run object.
+        """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not assistant_id:
             raise InputRequired("assistant_id is required")
         data = {"assistant_id": assistant_id}
@@ -140,6 +182,11 @@ def create(  # pylint: disable=too-many-branches
         Returns:
             Run: The `Run` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id or not assistant_id:
             raise InputRequired("thread_id and assistant_id is required")
         data = {"assistant_id": assistant_id}
@@ -309,6 +356,11 @@ def list(  # type: ignore[override]
         Returns:
             RunList: The list of runs.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id:
             raise InputRequired("thread_id is required!")
         response = super().list(
@@ -345,6 +397,11 @@ def retrieve(
         Returns:
             Run: The `Run` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id or not run_id:
             raise InputRequired("thread_id and run_id are required!")
         response = super().get(
@@ -399,7 +456,7 @@ def submit_tool_outputs(
         api_key: str = None,
         **kwargs,
     ) -> Run:
-        """_summary_
+        """Submit tool outputs.
 
         Args:
             thread_id (str): The thread id.
@@ -415,6 +472,11 @@ def submit_tool_outputs(
         Returns:
             Run: The 'Run`.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not tool_outputs:
             raise InputRequired("tool_outputs is required!")
         if not thread_id or not run_id:
@@ -568,6 +630,11 @@ def cancel(  # type: ignore[override]
         Returns:
             Run: The `Run` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id or not run_id:
             raise InputRequired("thread id and run id are required!")
         response = super().cancel(
diff --git a/dashscope/threads/runs/steps.py b/dashscope/threads/runs/steps.py
index 5299012..e7e0c86 100644
--- a/dashscope/threads/runs/steps.py
+++ b/dashscope/threads/runs/steps.py
@@ -1,14 +1,32 @@
 # -*- coding: utf-8 -*-
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
+import warnings
+
 from dashscope.client.base_api import GetStatusMixin, ListObjectMixin
 from dashscope.common.error import InputRequired
 from dashscope.threads.thread_types import RunStep, RunStepList
 
 __all__ = ["Steps"]
 
+# Deprecation warning message
+_DEPRECATION_MSG = (
+    "The Assistants API (dashscope.threads) is deprecated and will be "
+    "removed in a future release. Please migrate to the Responses API. "
+    "See https://help.aliyun.com/zh/model-studio/"
+    "synchronous-call-api-reference for migration details."
+)
+
 
 class Steps(ListObjectMixin, GetStatusMixin):
+    """
+    .. deprecated::
+        The Steps API (Assistants API) is deprecated and will be removed
+        in a future release. Please migrate to the Responses API.
+        See https://help.aliyun.com/zh/model-studio/
+        synchronous-call-api-reference for migration details.
+    """
+
     SUB_PATH = "RUNS"  # useless
 
     @classmethod
@@ -42,6 +60,11 @@ def list(  # type: ignore[override]
         Returns:
             RunList: The list of runs.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not run_id:
             raise InputRequired("run_id is required!")
         response = super().list(
@@ -80,6 +103,11 @@ def retrieve(
         Returns:
             RunStep: The `RunStep` object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id or not run_id or not step_id:
             raise InputRequired("thread_id, run_id and step_id are required!")
         response = super().get(
@@ -117,9 +145,9 @@ def get(  # type: ignore[override]
         """
         # pylint: disable=too-many-function-args
         return cls.retrieve(  # type: ignore[misc]
-            thread_id,
-            run_id,
             step_id,
+            thread_id=thread_id,
+            run_id=run_id,
             workspace=workspace,
             api_key=api_key,
             **kwargs,
diff --git a/dashscope/threads/threads.py b/dashscope/threads/threads.py
index 066232d..e4effe3 100644
--- a/dashscope/threads/threads.py
+++ b/dashscope/threads/threads.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
+import warnings
 from typing import Dict, List, Optional
 
 from dashscope.assistants.assistant_types import DeleteResponse
@@ -15,8 +16,23 @@
 
 __all__ = ["Threads"]
 
+_DEPRECATION_MSG = (
+    "The Assistants API (dashscope.threads) is deprecated and will be "
+    "removed in a future release. Please migrate to the Responses API. "
+    "See https://help.aliyun.com/zh/model-studio/"
+    "synchronous-call-api-reference for migration details."
+)
+
 
 class Threads(CreateMixin, DeleteMixin, GetStatusMixin, UpdateMixin):
+    """
+    .. deprecated::
+        The Threads API is deprecated and will be removed in a future release.
+        Please migrate to the Responses API.
+        See https://help.aliyun.com/zh/model-studio/
+        synchronous-call-api-reference for migration details.
+    """
+
     SUB_PATH = "threads"
 
     @classmethod
@@ -77,6 +93,11 @@ def create(
         Returns:
             Thread: The thread object.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         data = {}
         if messages:
             data["messages"] = messages
@@ -138,6 +159,11 @@ def retrieve(
         Returns:
             Thread: The `Thread` information.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id:
             raise InputRequired("thread_id is required!")
         response = super().get(
@@ -172,6 +198,11 @@ def update(  # type: ignore[override]
         Returns:
             Thread: The `Thread` information.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id:
             raise InputRequired("thread_id is required!")
         response = super().update(
@@ -205,6 +236,11 @@ def delete(  # type: ignore[override]
         Returns:
             AssistantsDeleteResponse: The deleted information.
         """
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not thread_id:
             raise InputRequired("thread_id is required!")
         response = super().delete(
@@ -231,6 +267,11 @@ def create_and_run(
         api_key: str = None,
         **kwargs,
     ) -> Run:
+        warnings.warn(
+            _DEPRECATION_MSG,
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         if not assistant_id:
             raise InputRequired("assistant_id is required")
         data = {"assistant_id": assistant_id}
diff --git a/dashscope/version.py b/dashscope/version.py
index 5aab490..0765d7e 100644
--- a/dashscope/version.py
+++ b/dashscope/version.py
@@ -1,4 +1,4 @@
 # -*- coding: utf-8 -*-
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
-__version__ = "1.26.4"
+__version__ = "1.26.5"
diff --git a/tests/unit/test_sync_custom_session.py b/tests/unit/test_sync_custom_session.py
index 7b493f3..6198293 100644
--- a/tests/unit/test_sync_custom_session.py
+++ b/tests/unit/test_sync_custom_session.py
@@ -119,12 +119,12 @@ def test_custom_session_is_used_for_request(self, _mock_session_class):
         # 验证自定义 session 没有被关闭
         mock_session.close.assert_not_called()
 
-    @patch("requests.Session")
-    def test_temporary_session_is_created_when_no_custom_session(
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
+    def test_shared_session_is_used_when_no_custom_session(
         self,
-        mock_session_class,
+        mock_get_session,
     ):
-        """测试没有自定义 session 时会创建临时 session"""
+        """测试没有自定义 session 时使用共享 session"""
         # 创建 mock session
         mock_session = Mock()
         mock_response = Mock()
@@ -132,7 +132,7 @@ def test_temporary_session_is_created_when_no_custom_session(
         mock_response.headers = {"content-type": "application/json"}
         mock_response.text = '{"status": "success"}'
         mock_session.post.return_value = mock_response
-        mock_session_class.return_value = mock_session
+        mock_get_session.return_value = mock_session
 
         # 创建 HttpRequest 不传 session
         http_request = HttpRequest(
@@ -163,11 +163,11 @@ def test_temporary_session_is_created_when_no_custom_session(
         ):
             _ = http_request.call()
 
-        # 验证临时 session 被创建
-        mock_session_class.assert_called_once()
+        # 验证共享 session 被使用
+        mock_get_session.assert_called_once()
 
-        # 验证临时 session 被关闭
-        mock_session.close.assert_called_once()
+        # 验证共享 session 没有被关闭
+        mock_session.close.assert_not_called()
 
 
 class TestSyncSessionResourceManagement:
@@ -212,16 +212,16 @@ def test_custom_session_not_closed_by_http_request(self):
         # 验证自定义 session 没有被关闭
         custom_session.close.assert_not_called()
 
-    @patch("requests.Session")
-    def test_temporary_session_closed_on_success(self, mock_session_class):
-        """测试临时 session 在成功后被关闭"""
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
+    def test_shared_session_not_closed_on_success(self, mock_get_session):
+        """测试共享 session 在成功后不被关闭"""
         mock_session = Mock()
         mock_response = Mock()
         mock_response.status_code = 200
         mock_response.headers = {"content-type": "application/json"}
         mock_response.text = '{"status": "success"}'
         mock_session.post.return_value = mock_response
-        mock_session_class.return_value = mock_session
+        mock_get_session.return_value = mock_session
 
         http_request = HttpRequest(
             url="http://example.com/api",
@@ -249,15 +249,15 @@ def test_temporary_session_closed_on_success(self, mock_session_class):
         ):
             _ = http_request.call()
 
-        # 验证临时 session 被关闭
-        mock_session.close.assert_called_once()
+        # 验证共享 session 没有被关闭
+        mock_session.close.assert_not_called()
 
-    @patch("requests.Session")
-    def test_temporary_session_closed_on_exception(self, mock_session_class):
-        """测试临时 session 在异常时也被关闭"""
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
+    def test_shared_session_not_closed_on_exception(self, mock_get_session):
+        """测试共享 session 在异常时不被关闭"""
         mock_session = Mock()
         mock_session.post.side_effect = Exception("Network error")
-        mock_session_class.return_value = mock_session
+        mock_get_session.return_value = mock_session
 
         http_request = HttpRequest(
             url="http://example.com/api",
@@ -282,8 +282,8 @@ def test_temporary_session_closed_on_exception(self, mock_session_class):
         with pytest.raises(Exception, match="Network error"):
             _ = http_request.call()
 
-        # 验证临时 session 仍然被关闭
-        mock_session.close.assert_called_once()
+        # 验证共享 session 没有被关闭
+        mock_session.close.assert_not_called()
 
 
 class TestSyncSessionWithCustomConfiguration:
@@ -472,16 +472,16 @@ def test_works_without_session_parameter(self):
         assert http_request.url == "http://example.com/api"
         assert http_request.method == HTTPMethod.POST
 
-    @patch("requests.Session")
-    def test_default_behavior_unchanged(self, mock_session_class):
-        """测试默认行为未改变"""
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
+    def test_default_behavior_unchanged(self, mock_get_session):
+        """测试默认行为：使用共享 session"""
         mock_session = Mock()
         mock_response = Mock()
         mock_response.status_code = 200
         mock_response.headers = {"content-type": "application/json"}
         mock_response.text = '{"status": "success"}'
         mock_session.post.return_value = mock_response
-        mock_session_class.return_value = mock_session
+        mock_get_session.return_value = mock_session
 
         # 不传 session 参数
         http_request = HttpRequest(
@@ -510,6 +510,7 @@ def test_default_behavior_unchanged(self, mock_session_class):
         ):
             _ = http_request.call()
 
-        # 验证临时 session 被创建和关闭（原有行为）
-        mock_session_class.assert_called_once()
-        mock_session.close.assert_called_once()
+        # 验证共享 session 被使用
+        mock_get_session.assert_called_once()
+        # 验证共享 session 没有被关闭
+        mock_session.close.assert_not_called()
diff --git a/tests/unit/test_utf8_encoding.py b/tests/unit/test_utf8_encoding.py
index ebee93c..32dcaf2 100644
--- a/tests/unit/test_utf8_encoding.py
+++ b/tests/unit/test_utf8_encoding.py
@@ -59,6 +59,7 @@ def _make_mock_sync_session():
 class _MockAioResponse:
     status = 200
     content_type = "application/json"
+    headers = {"Content-Type": "application/json"}
 
     async def json(self):
         return {"output": {"text": "ok"}, "request_id": "test-123"}
@@ -161,7 +162,7 @@ def test_accept_has_charset(self):
 class TestHttpRequestUtf8Body:
     # pylint: disable=protected-access
 
-    @patch("dashscope.api_entities.http_request.requests.Session")
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
     def test_body_is_utf8_bytes(self, mock_session_cls):
         mock_session = _make_mock_sync_session()
         mock_session_cls.return_value = mock_session
@@ -185,7 +186,7 @@ def test_body_is_utf8_bytes(self, mock_session_cls):
         assert isinstance(sent_body, bytes), "Body must be bytes, not str"
         assert sent_body == EXPECTED_BODY
 
-    @patch("dashscope.api_entities.http_request.requests.Session")
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
     def test_no_unicode_escapes_in_body(self, mock_session_cls):
         mock_session = _make_mock_sync_session()
         mock_session_cls.return_value = mock_session
@@ -211,7 +212,7 @@ def test_no_unicode_escapes_in_body(self, mock_session_cls):
         assert "Оцени идею" in body_str
         assert "こんにちは" in body_str
 
-    @patch("dashscope.api_entities.http_request.requests.Session")
+    @patch("dashscope.api_entities.http_request._get_shared_sync_session")
     def test_body_is_smaller_than_ascii_escaped(self, mock_session_cls):
         mock_session = _make_mock_sync_session()
         mock_session_cls.return_value = mock_session