diff --git a/dashscope/api_entities/aiohttp_request.py b/dashscope/api_entities/aiohttp_request.py
index d220ffb..97d8d32 100644
--- a/dashscope/api_entities/aiohttp_request.py
+++ b/dashscope/api_entities/aiohttp_request.py
@@ -3,7 +3,7 @@
 
 import json
 from http import HTTPStatus
-from typing import Optional
+from typing import Optional, Dict
 
 import aiohttp
 
@@ -17,7 +17,11 @@
 )
 from dashscope.common.error import UnsupportedHTTPMethod
 from dashscope.common.logging import logger
-from dashscope.common.utils import async_to_sync
+from dashscope.common.error_registry import INTERNAL_ERROR
+from dashscope.common.utils import (
+    async_to_sync,
+    _handle_aiohttp_failed_response,
+)
 
 
 class AioHttpRequest(AioBaseRequest):
@@ -85,10 +89,10 @@ def __init__(
         else:
             self.timeout = timeout  # type: ignore[has-type]
 
-    def add_header(self, key, value):
+    def add_header(self, key: str, value: str) -> None:
         self.headers[key] = value
 
-    def add_headers(self, headers):
+    def add_headers(self, headers: Dict[str, str]) -> None:
         self.headers = {**self.headers, **headers}
 
     def call(self):
@@ -97,9 +101,8 @@ def call(self):
             return (item for item in response)
         else:
             output = next(response)
-            try:
-                next(response)
-            except StopIteration:
+            # Consume remaining items to ensure generator completes
+            for _ in response:
                 pass
             return output
 
@@ -109,9 +112,8 @@ async def aio_call(self):
             return (item async for item in response)
         else:
             result = await response.__anext__()
-            try:
-                await response.__anext__()
-            except StopAsyncIteration:
+            # Consume remaining items to ensure generator completes
+            async for _ in response:
                 pass
             return result
 
@@ -142,6 +144,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
@@ -162,19 +165,29 @@ async def _handle_response(  # pylint: disable=too-many-branches
                     if "request_id" in msg:
                         request_id = msg["request_id"]
                 except json.JSONDecodeError:
+                    logger.error(
+                        "Failed to parse SSE stream data: %s",
+                        data[:200] if len(data) > 200 else data,
+                    )
                     yield DashScopeAPIResponse(
                         request_id=request_id,
-                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                        code="Unknown",
-                        message=data,
+                        status_code=INTERNAL_ERROR.status_code,
+                        code=INTERNAL_ERROR.error_code,
+                        message=INTERNAL_ERROR.error_msg,
+                        headers=headers,
                     )
                     continue
-                if is_error:
+                if is_error and msg is not None:
                     yield DashScopeAPIResponse(
                         request_id=request_id,
                         status_code=status_code,
-                        code=msg["code"],
-                        message=msg["message"],
+                        code=msg.get("code")
+                        or msg.get("error_code")
+                        or f"http_{status_code}",
+                        message=msg.get("message")
+                        or msg.get("error_message")
+                        or f"HTTP {status_code} error",
+                        headers=headers,
                     )
                 else:
                     yield DashScopeAPIResponse(
@@ -182,6 +195,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 +215,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,51 +232,22 @@ 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:
-                error = await response.json()
-                if "request_id" in error:
-                    request_id = error["request_id"]
-                if "message" not in error:
-                    message = ""
-                    logger.error(
-                        "Request: %s failed, status: %s",
-                        self.url,
-                        response.status,
-                    )
-                else:
-                    message = error["message"]
-                    logger.error(
-                        "Request: %s failed, status: %s, message: %s",
-                        self.url,
-                        response.status,
-                        error["message"],
-                    )
-                yield DashScopeAPIResponse(
-                    request_id=request_id,
-                    status_code=response.status,
-                    code=error["code"],
-                    message=message,
-                )
-            else:
-                msg = await response.read()
-                yield DashScopeAPIResponse(
-                    request_id=request_id,
-                    status_code=response.status,
-                    code="Unknown",
-                    message=msg.decode("utf-8"),
-                )
+            yield _handle_aiohttp_failed_response(response)
 
     # pylint: disable=too-many-branches
     async def _handle_request(self):
         try:
+            # Session management:
+            # - External session: managed by caller, we never close it
+            # - Shared session: managed by get_shared_aio_session(),
+            #   uses connection pooling and is closed when no longer needed
             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(
@@ -315,8 +301,21 @@ async def _handle_request(self):
                     async for rsp in self._handle_response(response):
                         yield rsp
             finally:
-                if should_close:
-                    await session.close()
+                # Note: We don't close the session here because:
+                # - External sessions are managed by the caller
+                # - Shared sessions use connection pooling and are
+                #   managed centrally by get_shared_aio_session()
+                pass
         except Exception as e:
-            logger.debug(e)
-            raise e
+            logger.error(
+                "Request failed: url=%s, method=%s, error=%s",
+                self.url,
+                self.method,
+                str(e),
+                exc_info=True,
+            )
+            from dashscope.common.error import DashScopeException
+
+            raise DashScopeException(
+                f"Request failed: {str(e)}",
+            ) from e
diff --git a/dashscope/api_entities/api_request_factory.py b/dashscope/api_entities/api_request_factory.py
index c231287..a9da3b9 100644
--- a/dashscope/api_entities/api_request_factory.py
+++ b/dashscope/api_entities/api_request_factory.py
@@ -69,6 +69,68 @@ def _get_protocol_params(kwargs):
     )
 
 
+def _build_http_url(
+    base_address: str,
+    is_service: bool,
+    task_group: str,
+    task: str,
+    function: str,
+    extra_url_parameters: dict,
+) -> str:
+    """Build HTTP/HTTPS URL from components with validation."""
+    if base_address is None:
+        base_address = dashscope.base_http_api_url
+
+    # Validate base_address has proper scheme (http:// or https://)
+    if base_address and not base_address.startswith(
+        ("http://", "https://"),
+    ):
+        from dashscope.common.error import InvalidBaseURL
+
+        raise InvalidBaseURL(
+            f"Invalid URL '{base_address}': No scheme supplied. "
+            f"Perhaps you meant https://{base_address}?",
+        )
+
+    if not base_address.endswith("/"):
+        http_url = base_address + "/"
+    else:
+        http_url = base_address
+
+    if is_service:
+        http_url = http_url + SERVICE_API_PATH + "/"
+
+    if task_group:
+        http_url += f"{task_group}/"
+    if task:
+        http_url += f"{task}/"
+    if function:
+        http_url += function
+    if extra_url_parameters is not None and extra_url_parameters:
+        http_url += "?" + urlencode(extra_url_parameters)
+
+    return http_url
+
+
+def _build_websocket_url(base_address: str) -> str:
+    """Build and validate WebSocket URL."""
+    if base_address is not None:
+        websocket_url = base_address
+    else:
+        websocket_url = dashscope.base_websocket_api_url
+
+    # Validate websocket_url has proper scheme (ws:// or wss://)
+    if websocket_url and not websocket_url.startswith(("ws://", "wss://")):
+        from dashscope.common.error import InvalidBaseURL
+
+        raise InvalidBaseURL(
+            f"Invalid URL '{websocket_url}': No scheme supplied. "
+            f"Perhaps you meant wss://{websocket_url}?",
+        )
+
+    return websocket_url
+
+
 def _build_api_request(  # pylint: disable=too-many-branches
     model: str,
     input: object,  # pylint: disable=redefined-builtin
@@ -102,24 +164,14 @@ def _build_api_request(  # pylint: disable=too-many-branches
     encryption = None
 
     if api_protocol in [ApiProtocol.HTTP, ApiProtocol.HTTPS]:
-        if base_address is None:
-            base_address = dashscope.base_http_api_url
-        if not base_address.endswith("/"):
-            http_url = base_address + "/"
-        else:
-            http_url = base_address
-
-        if is_service:
-            http_url = http_url + SERVICE_API_PATH + "/"
-
-        if task_group:
-            http_url += f"{task_group}/"
-        if task:
-            http_url += f"{task}/"
-        if function:
-            http_url += function
-        if extra_url_parameters is not None and extra_url_parameters:
-            http_url += "?" + urlencode(extra_url_parameters)
+        http_url = _build_http_url(
+            base_address,
+            is_service,
+            task_group,
+            task,
+            function,
+            extra_url_parameters,
+        )
 
         if enable_encryption is True:
             encryption = Encryption()
@@ -142,10 +194,7 @@ def _build_api_request(  # pylint: disable=too-many-branches
             session=session,
         )
     elif api_protocol == ApiProtocol.WEBSOCKET:
-        if base_address is not None:
-            websocket_url = base_address
-        else:
-            websocket_url = dashscope.base_websocket_api_url
+        websocket_url = _build_websocket_url(base_address)
         pre_task_id = kwargs.pop("pre_task_id", None)
         request = WebSocketRequest(
             url=websocket_url,
diff --git a/dashscope/api_entities/http_request.py b/dashscope/api_entities/http_request.py
index 285c6ab..72366fd 100644
--- a/dashscope/api_entities/http_request.py
+++ b/dashscope/api_entities/http_request.py
@@ -17,6 +17,7 @@
     HTTPMethod,
 )
 from dashscope.common.error import UnsupportedHTTPMethod
+from dashscope.common.error_registry import MISSING_PARAMETER, INVALID_REQUEST
 from dashscope.common.logging import logger
 from dashscope.common.utils import (
     _handle_aio_stream,
@@ -130,10 +131,10 @@ def __init__(
         else:
             self.timeout = timeout  # type: ignore[has-type]
 
-    def add_header(self, key, value):
+    def add_header(self, key: str, value: str) -> None:
         self.headers[key] = value
 
-    def add_headers(self, headers):
+    def add_headers(self, headers: Dict[str, str]) -> None:
         self.headers = {**self.headers, **headers}
 
     def call(self):
@@ -142,9 +143,8 @@ def call(self):
             return (item for item in response)
         else:
             output = next(response)
-            try:
-                next(response)
-            except StopIteration:
+            # Consume remaining items to ensure generator completes
+            for _ in response:
                 pass
             return output
 
@@ -154,9 +154,8 @@ async def aio_call(self):
             return (item async for item in response)
         else:
             result = await response.__anext__()
-            try:
-                await response.__anext__()
-            except StopAsyncIteration:
+            # Consume remaining items to ensure generator completes
+            async for _ in response:
                 pass
             return result
 
@@ -228,8 +227,10 @@ async def _handle_aio_request(self):  # pylint: disable=too-many-branches
                 if should_close:
                     await session.close()
         except Exception as e:
-            logger.debug(e)
-            raise e
+            logger.error(f"Async request failed: {e}", exc_info=True)
+            from dashscope.common.error import DashScopeException
+
+            raise DashScopeException(str(e)) from e
 
     @staticmethod
     def __handle_parameters(params: dict) -> dict:
@@ -290,9 +291,11 @@ async def _handle_aio_response(  # pylint: disable=too-many-branches, too-many-s
                 except json.JSONDecodeError:
                     yield DashScopeAPIResponse(
                         request_id=request_id,
-                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                        code="Unknown",
-                        message=data,
+                        status_code=MISSING_PARAMETER.status_code,
+                        code=MISSING_PARAMETER.error_code,
+                        message=MISSING_PARAMETER.format_msg(
+                            {"parameter": "response data"},
+                        ),
                         headers=headers,
                     )
                     continue
@@ -300,8 +303,12 @@ async def _handle_aio_response(  # pylint: disable=too-many-branches, too-many-s
                     yield DashScopeAPIResponse(
                         request_id=request_id,
                         status_code=status_code,
-                        code=msg["code"],
-                        message=msg["message"],
+                        code=msg.get("code")
+                        or msg.get("error_code")
+                        or f"http_{status_code}",
+                        message=msg.get("message")
+                        or msg.get("error_message")
+                        or f"HTTP {status_code} error",
                         headers=headers,
                     )
                 else:
@@ -399,10 +406,10 @@ def _handle_response(  # pylint: disable=too-many-branches
                 except json.JSONDecodeError:
                     yield DashScopeAPIResponse(
                         request_id=request_id,
-                        status_code=HTTPStatus.BAD_REQUEST,
+                        status_code=INVALID_REQUEST.status_code,
                         output=None,
-                        code="Unknown",
-                        message=data,
+                        code=INVALID_REQUEST.error_code,
+                        message=INVALID_REQUEST.error_msg,
                         headers=headers,
                     )
                     continue
@@ -411,10 +418,12 @@ def _handle_response(  # pylint: disable=too-many-branches
                         request_id=request_id,
                         status_code=status_code,
                         output=None,
-                        code=msg["code"]
-                        if "code" in msg
-                        else None,  # noqa E501
-                        message=msg["message"] if "message" in msg else None,
+                        code=msg.get("code")
+                        or msg.get("error_code")
+                        or f"http_{status_code}",
+                        message=msg.get("message")
+                        or msg.get("error_message")
+                        or f"HTTP {status_code} error",
                         headers=headers,
                     )  # noqa E501
                 else:
@@ -517,5 +526,7 @@ def _handle_request(self):  # pylint: disable=too-many-branches
                 if should_close:
                     session.close()
         except Exception as e:
-            logger.debug(e)
-            raise e
+            logger.error(f"Sync request failed: {e}", exc_info=True)
+            from dashscope.common.error import DashScopeException
+
+            raise DashScopeException(str(e)) from e
diff --git a/dashscope/api_entities/websocket_request.py b/dashscope/api_entities/websocket_request.py
index a3ceb97..0f0a704 100644
--- a/dashscope/api_entities/websocket_request.py
+++ b/dashscope/api_entities/websocket_request.py
@@ -5,7 +5,7 @@
 import json
 import uuid
 from http import HTTPStatus
-from typing import Tuple, Union
+from typing import Tuple, Union, Dict
 
 import aiohttp
 
@@ -13,7 +13,6 @@
 from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
 from dashscope.common.constants import (
     DEFAULT_REQUEST_TIMEOUT_SECONDS,
-    SERVICE_503_MESSAGE,
     WEBSOCKET_ERROR_CODE,
 )
 from dashscope.common.error import (
@@ -21,6 +20,11 @@
     UnexpectedMessageReceived,
     UnknownMessageReceived,
 )
+from dashscope.common.error_registry import (
+    SERVICE_UNAVAILABLE,
+    AUTH_FAILED,
+    INTERNAL_ERROR,
+)
 from dashscope.common.logging import logger
 from dashscope.common.utils import async_to_sync
 from dashscope.protocol.websocket import (
@@ -81,7 +85,7 @@ def __init__(
         }
         self.pre_task_id = pre_task_id
 
-    def add_headers(self, headers):
+    def add_headers(self, headers: Dict[str, str]) -> None:
         self.headers = {**self.headers, **headers}
 
     def call(self):
@@ -112,7 +116,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(
@@ -195,34 +201,67 @@ async def connection_handler(self):  # pylint: disable=too-many-branches
             )
         except aiohttp.ClientConnectorError as e:
             logger.exception(e)
+            # Check if the error message indicates service unavailability
+            error_str = str(e).lower()
+            if "service unavailable" in error_str or "503" in error_str:
+                yield DashScopeAPIResponse(
+                    request_id=task_id if task_id else "",
+                    status_code=SERVICE_UNAVAILABLE.status_code,
+                    code=SERVICE_UNAVAILABLE.error_code,
+                    message=SERVICE_UNAVAILABLE.error_msg,
+                )
+                return
+
             yield DashScopeAPIResponse(
-                request_id="",
-                status_code=-1,
-                code="ClientConnectorError",
-                message=str(e),
+                request_id=task_id if task_id else "",
+                status_code=INTERNAL_ERROR.status_code,
+                code=INTERNAL_ERROR.error_code,
+                message=INTERNAL_ERROR.error_msg,
             )
         except aiohttp.WSServerHandshakeError as e:
-            code = e.status
-            msg = e.message
+            original_msg = e.message or ""
+
             if e.status in [HTTPStatus.FORBIDDEN, HTTPStatus.UNAUTHORIZED]:
-                msg = "Unauthorized, your api-key is invalid!"
+                yield DashScopeAPIResponse(
+                    request_id=task_id if task_id else "",
+                    status_code=AUTH_FAILED.status_code,
+                    code=AUTH_FAILED.error_code,
+                    message=AUTH_FAILED.error_msg,
+                )
             elif e.status == HTTPStatus.SERVICE_UNAVAILABLE:
-                msg = SERVICE_503_MESSAGE
+                yield DashScopeAPIResponse(
+                    request_id=task_id if task_id else "",
+                    status_code=SERVICE_UNAVAILABLE.status_code,
+                    code=SERVICE_UNAVAILABLE.error_code,
+                    message=SERVICE_UNAVAILABLE.error_msg,
+                )
             else:
-                pass
-            yield DashScopeAPIResponse(
-                request_id=task_id,
-                status_code=code,
-                code=code,
-                message=msg,
-            )
-        except BaseException as e:
+                if e.status == HTTPStatus.INTERNAL_SERVER_ERROR:
+                    yield DashScopeAPIResponse(
+                        request_id=task_id if task_id else "",
+                        status_code=INTERNAL_ERROR.status_code,
+                        code=INTERNAL_ERROR.error_code,
+                        message=INTERNAL_ERROR.error_msg,
+                    )
+                else:
+                    # Log unexpected status codes for debugging
+                    logger.warning(
+                        "WebSocket handshake failed with unexpected "
+                        "status %s: %s",
+                        e.status,
+                        original_msg,
+                    )
+                    raise e
+        except Exception as e:
             logger.exception(e)
             yield DashScopeAPIResponse(
-                request_id="",
-                status_code=-1,
-                code="Unknown",
-                message=f"Error type: {type(e)}, message: {e}",
+                request_id=task_id if task_id else "",
+                status_code=INTERNAL_ERROR.status_code,
+                code=INTERNAL_ERROR.error_code,
+                message=(
+                    f"{INTERNAL_ERROR.error_msg} "
+                    f"(SDK Internal Error: {type(e).__name__}: {e})"
+                ),
             )
 
     def _to_DashScopeAPIResponse(self, task_id, is_binary, result):
diff --git a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
index 3d5c374..2a7e077 100644
--- a/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
+++ b/dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py
@@ -10,7 +10,7 @@
 import websocket  # pylint: disable=wrong-import-order
 
 import dashscope
-from dashscope.common.error import ModelRequired
+from dashscope.common.error import AuthenticationError, ModelRequired
 from dashscope.common.logging import logger
 from dashscope.common.utils import get_user_agent
 
@@ -105,6 +105,8 @@ def __init__(
         self.last_first_text_time = None
         self.last_first_audio_delay = None
         self.metrics = []
+        self._auth_failed = False  # Flag to track authentication failures
+        self._connection_error = None  # Store connection error details
 
     def _generate_event_id(self):
         """
@@ -131,6 +133,10 @@ def connect(self) -> None:
         """
         connect to server, create session and return default session configuration  # noqa: E501
         """
+        # Reset error flags before connecting
+        self._auth_failed = False
+        self._connection_error = None
+
         self.ws = websocket.WebSocketApp(
             self.url,
             header=self._get_websocket_header(),
@@ -147,7 +153,16 @@ def connect(self) -> None:
             not (self.ws.sock and self.ws.sock.connected)
             and (time.time() - start_time) < timeout
         ):
+            # Check for authentication errors first
+            if self._auth_failed and self._connection_error is not None:
+                raise self._connection_error
+
             time.sleep(0.1)  # Brief sleep to avoid busy polling
+
+        # Final check after timeout
+        if self._auth_failed and self._connection_error is not None:
+            raise self._connection_error
+
         if not (self.ws.sock and self.ws.sock.connected):
             raise TimeoutError(
                 "websocket connection could not established within 5s. "
@@ -415,6 +430,20 @@ def on_close(  # pylint: disable=unused-argument
     # Callback for WebSocket error
     def on_error(self, ws, error):  # pylint: disable=unused-argument
         logger.error(f"websocket error: {error}")
+
+        # Detect authentication errors from error message
+        error_str = str(error)
+        if (
+            "401" in error_str
+            or "Unauthorized" in error_str
+            or "InvalidApiKey" in error_str
+        ):
+            self._auth_failed = True
+            self._connection_error = AuthenticationError(
+                "API Key is invalid. Please check your API key configuration.",
+            )
+            logger.error("Authentication failed: Invalid API Key")
+
         # Do not raise exception here, let the connection close naturally
         # Raising exception in callback can cause unexpected thread termination
 
diff --git a/dashscope/cli/agentic_rl.py b/dashscope/cli/agentic_rl.py
index cd8386c..65376c8 100644
--- a/dashscope/cli/agentic_rl.py
+++ b/dashscope/cli/agentic_rl.py
@@ -24,7 +24,11 @@
 )
 from dashscope.finetune.reinforcement.common.errors import OutputError
 from dashscope.finetune.customize_types import FineTune
-from dashscope.cli.common import error, normalize_local_path_or_url
+from dashscope.cli.common import (
+    error,
+    normalize_local_path_or_url,
+    _handle_exception,
+)
 
 
 app = typer.Typer(
@@ -215,9 +219,8 @@ async def _register_fc_async(
         }
 
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ FC registration failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "FC registration failed", err_console)
+        raise  # pragma: no cover
 
 
 @app.command("register-functions", hidden=True)
@@ -292,9 +295,8 @@ async def _test_fc_async(
         return result
 
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ Function test failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "Function test failed", err_console)
+        raise  # pragma: no cover
 
 
 @app.command("test-functions", hidden=True)
@@ -370,9 +372,7 @@ async def _upload_data_async(
         }
 
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ Dataset upload failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "Dataset upload failed", err_console)
 
 
 @app.command("upload-data", hidden=True)
@@ -536,8 +536,16 @@ def run(
             # Handle API response errors
             if result.status_code != 200:
                 raise OutputError(
-                    f"API returned status {result.status_code}:"
-                    f" {result.message}",
+                    (
+                        f"API error [status={result.status_code}, "
+                        f"code={result.code}]: {result.message}"
+                    ),
+                    response={
+                        "status_code": result.status_code,
+                        "code": result.code,
+                        "message": result.message,
+                        "request_id": result.request_id,
+                    },
                 )
 
             progress.update(
@@ -559,24 +567,22 @@ def run(
         )
 
     except Exception as e:
-        root = _root_cause(e)
         label = (
             "Validation error"
             if isinstance(e, ValueError)
             else "Workflow execution failed"
         )
-        err_console.print(f"[red]❌ {label}: {root}[/red]")
+        _handle_exception(e, label, err_console)
         if _cli_verbose:
             err_console.print(
                 "".join(
                     tb_module.format_exception(
-                        type(root),
-                        root,
-                        root.__traceback__,
+                        type(e),
+                        e,
+                        e.__traceback__,
                     ),
                 ),
             )
-        raise typer.Exit(1)
 
 
 @app.command()
@@ -598,7 +604,16 @@ def get(
         # Handle API response errors
         if result.status_code != 200:
             raise OutputError(
-                f"API returned status {result.status_code}: {result.message}",
+                (
+                    f"API error [status={result.status_code}, "
+                    f"code={result.code}]: {result.message}"
+                ),
+                response={
+                    "status_code": result.status_code,
+                    "code": result.code,
+                    "message": result.message,
+                    "request_id": result.request_id,
+                },
             )
 
         # Validate output is not None before accessing attributes
@@ -614,9 +629,7 @@ def get(
             fmt=output_format,
         )
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ Query failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "Query failed", err_console)
 
 
 @app.command()
@@ -636,16 +649,23 @@ def cancel(
         # Handle API response errors
         if result.status_code != 200:
             raise OutputError(
-                f"API returned status {result.status_code}: {result.message}",
+                (
+                    f"API error [status={result.status_code}, "
+                    f"code={result.code}]: {result.message}"
+                ),
+                response={
+                    "status_code": result.status_code,
+                    "code": result.code,
+                    "message": result.message,
+                    "request_id": result.request_id,
+                },
             )
 
         err_console.print(
             f"[green]✅ Job {job_id} canceled successfully[/green]",
         )
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ Cancellation failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "Cancellation failed", err_console)
 
 
 @app.command()
@@ -674,22 +694,32 @@ def logs(
         # Handle API response errors
         if result.status_code != 200:
             raise OutputError(
-                f"API returned status {result.status_code}: {result.message}",
+                (
+                    f"API error [status={result.status_code}, "
+                    f"code={result.code}]: {result.message}"
+                ),
+                response={
+                    "status_code": result.status_code,
+                    "code": result.code,
+                    "message": result.message,
+                    "request_id": result.request_id,
+                },
             )
 
         # Validate output is not None before accessing attributes
         logs_data = ""
-        if result.output is not None and isinstance(result.output, dict):
-            logs_data = result.output.get("logs", "")
+        if result.output is not None:
+            if isinstance(result.output, dict):
+                logs_data = result.output.get("logs", "")
+            else:
+                logs_data = getattr(result.output, "logs", "")
 
         format_output(
             {"job_id": job_id, "logs": logs_data},
             fmt=output_format,
         )
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ Log retrieval failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "Log retrieval failed", err_console)
 
 
 @app.command("list")
@@ -716,7 +746,16 @@ def list_jobs(
         # Handle API response errors
         if result.status_code != 200:
             raise OutputError(
-                f"API returned status {result.status_code}: {result.message}",
+                (
+                    f"API error [status={result.status_code}, "
+                    f"code={result.code}]: {result.message}"
+                ),
+                response={
+                    "status_code": result.status_code,
+                    "code": result.code,
+                    "message": result.message,
+                    "request_id": result.request_id,
+                },
             )
 
         output_data = serialize_for_output(
@@ -724,9 +763,7 @@ def list_jobs(
         )
         format_output(output_data, fmt=output_format)
     except Exception as e:
-        root = _root_cause(e)
-        err_console.print(f"[red]❌ List query failed: {root}[/red]")
-        raise typer.Exit(1)
+        _handle_exception(e, "List query failed", err_console)
 
 
 # if __name__ == "__main__":
diff --git a/dashscope/cli/common.py b/dashscope/cli/common.py
index ba7fb3f..ae9719c 100644
--- a/dashscope/cli/common.py
+++ b/dashscope/cli/common.py
@@ -4,12 +4,14 @@
 import os
 from functools import wraps
 from http import HTTPStatus
-from typing import Callable, NoReturn, TypeVar
+from typing import Callable, Dict, NoReturn, TypeVar
 from urllib.parse import urlparse
 
 import typer
 from rich.console import Console
 
+from dashscope.common.error import DashScopeException
+
 logger = logging.getLogger("dashscope.cli")
 CommandFunction = TypeVar("CommandFunction", bound=Callable)
 
@@ -21,6 +23,123 @@
 DEFAULT_PAGE_SIZE = 10
 DEFAULT_START_PAGE = 1
 
+# ---------------------------------------------------------------------------
+# Error code mapping: SDK error codes -> CLI-friendly error codes
+# ---------------------------------------------------------------------------
+ERROR_CODE_MAPPING: Dict[str, str] = {
+    # Authentication errors - keep original camelCase format
+    "AuthenticationError": "AuthenticationError",
+    "AuthFailed": "AuthFailed",
+    "InvalidToken": "InvalidToken",
+    "TokenExpired": "TokenExpired",
+    "Unauthorized": "Unauthorized",
+    # Parameter errors - keep original camelCase format
+    "InvalidParameter": "InvalidParameter",
+    "InvalidParam": "InvalidParam",
+    "BadRequest": "BadRequest",
+    "ModelRequired": "ModelRequired",
+    "InvalidModel": "InvalidModel",
+    "InvalidInput": "InvalidInput",
+    "InvalidFileFormat": "InvalidFileFormat",
+    "InputDataRequired": "InputDataRequired",
+    "InputRequired": "InputRequired",
+    "UnsupportedDataType": "UnsupportedDataType",
+    # Task errors - keep original camelCase format
+    "InvalidTask": "InvalidTask",
+    "UnsupportedTask": "UnsupportedTask",
+    "UnsupportedModel": "UnsupportedModel",
+    "UnsupportedApiProtocol": "UnsupportedApiProtocol",
+    "NotImplemented": "NotImplemented",
+    "MultiInputsWithBinaryNotSupported": "MultiInputsWithBinaryNotSupported",
+    "UnexpectedMessageReceived": "UnexpectedMessageReceived",
+    "UnsupportedData": "UnsupportedData",
+    "UnknownMessageReceived": "UnknownMessageReceived",
+    # Service errors - keep original camelCase format
+    "ServiceUnavailableError": "ServiceUnavailableError",
+    "UnsupportedHTTPMethod": "UnsupportedHTTPMethod",
+    "AsyncTaskCreateFailed": "AsyncTaskCreateFailed",
+    "UploadFileException": "UploadFileException",
+    "TimeoutException": "TimeoutException",
+    # Assistant errors - keep original camelCase format
+    "AssistantError": "AssistantError",
+}
+
+# Error message templates with CLI context
+ERROR_MESSAGE_TEMPLATES: Dict[str, str] = {
+    "AuthFailed": (
+        "Authentication failed. " "Please check your API key and try again."
+    ),
+    "InvalidToken": (
+        "Authentication failed. " "Please check your API key and try again."
+    ),
+    "TokenExpired": (
+        "Authentication failed. " "Please check your API key and try again."
+    ),
+    "Unauthorized": (
+        "Authentication failed. " "Please check your API key and try again."
+    ),
+    "InvalidParameter": (
+        "Invalid parameter provided. " "Please check your input parameters."
+    ),
+    "InvalidParam": (
+        "Invalid parameter provided. " "Please check your input parameters."
+    ),
+    "ModelRequired": (
+        "Model parameter is required. " "Please specify a valid model."
+    ),
+    "InvalidModel": (
+        "Invalid model specified. " "Please check the model name."
+    ),
+    "InvalidInput": ("Invalid input data. " "Please check your input format."),
+    "InvalidFileFormat": (
+        "Invalid file format. " "Please check the file type."
+    ),
+    "InputDataRequired": (
+        "Input data is required. " "Please provide the necessary input."
+    ),
+    "InputRequired": (
+        "Input is required. " "Please provide the necessary input."
+    ),
+    "UnsupportedDataType": (
+        "Unsupported data type. " "Please check the data format."
+    ),
+    "InvalidTask": ("Invalid task specified. " "Please check the task type."),
+    "UnsupportedTask": (
+        "Unsupported task type. " "Please check the available tasks."
+    ),
+    "UnsupportedModel": (
+        "Unsupported model. " "Please check the available models."
+    ),
+    "UnsupportedApiProtocol": (
+        "Unsupported API protocol. " "Please check the protocol version."
+    ),
+    "NotImplemented": "This feature is not yet implemented.",
+    "MultiInputsWithBinaryNotSupported": (
+        "Binary input is not supported with multiple inputs."
+    ),
+    "UnexpectedMessageReceived": (
+        "Unexpected message received from the server."
+    ),
+    "UnsupportedData": "Unsupported data format.",
+    "UnknownMessageReceived": ("Unknown message received from the server."),
+    "ServiceUnavailableError": (
+        "Service is temporarily unavailable. " "Please try again later."
+    ),
+    "UnsupportedHTTPMethod": (
+        "Unsupported HTTP method. " "Please check the request method."
+    ),
+    "AsyncTaskCreateFailed": (
+        "Failed to create async task. " "Please check your request."
+    ),
+    "UploadFileException": (
+        "File upload failed. " "Please check the file and try again."
+    ),
+    "TimeoutException": "Request timed out. Please try again.",
+    "AssistantError": (
+        "Assistant encountered an error. " "Please check the error details."
+    ),
+}
+
 # ---------------------------------------------------------------------------
 # Rich consoles
 # ---------------------------------------------------------------------------
@@ -28,20 +147,158 @@
 err_console = Console(stderr=True)
 
 # ---------------------------------------------------------------------------
-# Response helpers
+# Error handling utilities
 # ---------------------------------------------------------------------------
 
 
-def print_failed_message(rsp):
-    """Print a standardised error message for a failed API response."""
-    err_console.print(
-        f"[red]Failed[/red] request_id: {rsp.request_id}, "
-        f"status_code: {rsp.status_code}, "
-        f"code: {rsp.code}, message: {rsp.message}",
+def _get_cli_error_code(sdk_error_code: str) -> str:
+    """Map SDK error code to CLI-friendly error code.
+
+    Args:
+        sdk_error_code: The error code from the SDK response
+
+    Returns:
+        CLI-friendly error code, or the original code if no mapping exists
+    """
+    return ERROR_CODE_MAPPING.get(sdk_error_code, sdk_error_code)
+
+
+def _get_cli_error_message(sdk_error_code: str, sdk_error_message: str) -> str:
+    """Get CLI-friendly error message with context.
+
+    Args:
+        sdk_error_code: The error code from the SDK response
+        sdk_error_message: The error message from the SDK response
+
+    Returns:
+        CLI-friendly error message
+    """
+    cli_error_code = _get_cli_error_code(sdk_error_code)
+
+    # Use template message if available, otherwise use SDK message
+    template_message = ERROR_MESSAGE_TEMPLATES.get(cli_error_code)
+    if template_message:
+        # Append SDK-specific message if available
+        if sdk_error_message:
+            return f"{sdk_error_message}"
+        return template_message
+
+    # No template available, use SDK message directly
+    return (
+        sdk_error_message
+        if sdk_error_message
+        else f"Error code: {cli_error_code}"
     )
 
 
-def ensure_ok(rsp, check_business_error: bool = True):
+def _format_error_parts(
+    request_id: str,
+    status_code: str,
+    cli_error_code: str,
+    cli_error_message: str,
+    command_name: str = None,
+) -> str:
+    """Build formatted error output parts.
+
+    Args:
+        request_id: The request ID from the response
+        status_code: The HTTP status code
+        cli_error_code: The CLI-friendly error code
+        cli_error_message: The CLI-friendly error message
+        command_name: The CLI command name (optional)
+
+    Returns:
+        Formatted error message string
+    """
+    parts = []
+    if command_name:
+        parts.append(f"[red]{command_name} failed[/red]")
+    else:
+        parts.append("[red]Request failed[/red]")
+
+    if request_id and request_id != "N/A":
+        parts.append(f"request_id: {request_id}")
+    if status_code and status_code != "N/A":
+        parts.append(f"status_code: {status_code}")
+    parts.append(f"code: {cli_error_code}")
+    parts.append(f"message: {cli_error_message}")
+
+    return ", ".join(parts)
+
+
+# ---------------------------------------------------------------------------
+# Response helpers
+# ---------------------------------------------------------------------------
+
+
+def print_failed_message(rsp, command_name: str = None):
+    """Print a standardised error message for a failed API response.
+
+    Maps SDK error codes to CLI-friendly codes and enhances error messages
+    with CLI context. Safely handles responses with missing or None attributes.
+
+    Args:
+        rsp: The API response object
+        command_name: The CLI command name (optional, for better context)
+    """
+    # Use try-except to handle missing attributes gracefully (works with Mock
+    # objects)
+    try:
+        request_id = rsp.request_id
+    except AttributeError:
+        request_id = None
+
+    try:
+        status_code = rsp.status_code
+    except AttributeError:
+        status_code = None
+
+    try:
+        code = rsp.code
+    except AttributeError:
+        code = None
+
+    try:
+        message = rsp.message
+    except AttributeError:
+        message = None
+
+    # Normalize None and empty strings
+    request_id = request_id if request_id else "N/A"
+    status_code = status_code if status_code is not None else "N/A"
+    code = code if code else ""
+    message = message if message else ""
+
+    # Use the new error formatting with CLI context
+    if code:
+        cli_error_code = _get_cli_error_code(code)
+        cli_error_message = _get_cli_error_message(code, message)
+
+        formatted_error = _format_error_parts(
+            request_id=request_id,
+            status_code=status_code,
+            cli_error_code=cli_error_code,
+            cli_error_message=cli_error_message,
+            command_name=command_name,
+        )
+        err_console.print(formatted_error)
+    else:
+        # Fallback for responses without error code
+        parts = ["[red]Failed[/red]"]
+        if request_id != "N/A":
+            parts.append(f"request_id: {request_id}")
+        if status_code != "N/A":
+            parts.append(f"status_code: {status_code}")
+        if message:
+            parts.append(f"message: {message}")
+        err_console.print(", ".join(parts))
+
+
+def ensure_ok(
+    rsp,
+    check_business_error: bool = True,
+    command_name: str = None,
+):
     """Return *rsp.output* when the response is OK; otherwise print the error
     and exit with code 1.
 
@@ -50,7 +307,7 @@ def ensure_ok(rsp, check_business_error: bool = True):
 
     Enhanced to check both HTTP status and business-level error codes:
     - HTTP 200 but InvalidParameter → still treated as failure
-    - HTTP 4xx/5xx → clear error message
+    - HTTP 4xx/5xx → clear error message with CLI context
 
     Args:
         rsp: The API response object
@@ -58,15 +315,22 @@ def ensure_ok(rsp, check_business_error: bool = True):
                               error codes in the output. Set to False for
                               async task creation where we only care about
                               HTTP success, not task execution.
+        command_name: The CLI command name (optional, for better context)
     """
+    # Check HTTP status first
     if rsp.status_code != HTTPStatus.OK:
-        print_failed_message(rsp)
+        print_failed_message(rsp, command_name=command_name)
         raise typer.Exit(1)
 
-    # Check for business-level errors even when HTTP status is 200
+    # Check if output exists
     output = rsp.output
     if output is None:
-        print_failed_message(rsp)
+        # HTTP 200 but no output - this is unusual, treat as error
+        err_console.print(
+            f"[red]Error[/red] "
+            f"request_id: {getattr(rsp, 'request_id', 'N/A')}, "
+            f"HTTP 200 but response has no output data",
+        )
         raise typer.Exit(1)
 
     # Only check business-level errors if explicitly requested
@@ -74,18 +338,29 @@ def ensure_ok(rsp, check_business_error: bool = True):
         # Some APIs return error info in output even with HTTP 200
         if isinstance(output, dict):
             error_code = output.get("code")
-            message = output.get("message", "Unknown error")
+            message = output.get("message")
         else:
             error_code = getattr(output, "code", None)
-            message = getattr(output, "message", "Unknown error")
-
-        if error_code and error_code != "":
-            err_console.print(
-                f"[red]Business Error[/red] request_id: {rsp.request_id}, "
-                f"status_code: {rsp.status_code}, "
-                f"code: {error_code}, "
-                f"message: {message}",
+            message = getattr(output, "message", None)
+
+        # Only report if there's an actual error code
+        if error_code:
+            # Use the new error formatting with CLI context
+            request_id = getattr(rsp, "request_id", "N/A")
+            cli_error_code = _get_cli_error_code(error_code)
+            cli_error_message = _get_cli_error_message(
+                error_code,
+                message or "API returned error code without message",
             )
+
+            formatted_error = _format_error_parts(
+                request_id=request_id,
+                status_code=str(rsp.status_code),
+                cli_error_code=cli_error_code,
+                cli_error_message=cli_error_message,
+                command_name=command_name,
+            )
+            err_console.print(formatted_error)
             raise typer.Exit(1)
 
     return output
@@ -107,8 +382,64 @@ def error(message: str, exit_code: int = 1) -> NoReturn:
     raise typer.Exit(exit_code)
 
 
+def _handle_exception(
+    exception: Exception,
+    action: str,
+    output_console: Console,
+) -> NoReturn:
+    """Handle an exception and print a friendly error message.
+
+    Maps SDK exception types to CLI-friendly error codes and enhances
+    error messages with CLI context. Preserves full exception context
+    including stack trace for debugging.
+
+    Args:
+        exception: The exception to handle.
+        action: The action that failed (e.g., "FC registration failed").
+        output_console: The Rich console to print to.
+    """
+    if isinstance(exception, DashScopeException):
+        # Handle known DashScope exceptions with structured error info
+        request_id = getattr(exception, "request_id", "N/A") or "N/A"
+        message = getattr(exception, "message", str(exception))
+
+        # Map exception type to CLI-friendly error code
+        exception_type_name = type(exception).__name__
+        cli_error_code = _get_cli_error_code(exception_type_name)
+        cli_error_message = _get_cli_error_message(
+            exception_type_name,
+            message,
+        )
+
+        output_console.print(
+            f"[red]{action}[/red] "
+            f"(request_id: {request_id}, code: {cli_error_code})\n"
+            f"  {cli_error_message}",
+            no_wrap=True,
+        )
+        # Log full traceback for debugging
+        logger.debug(
+            f"{action} failed with DashScopeException",
+            exc_info=True,
+        )
+    else:
+        # Handle unexpected exceptions with full context
+        output_console.print(f"[red]{action}:[/red] {exception}")
+        # Log full traceback for debugging unexpected errors
+        logger.debug(
+            f"{action} failed with unexpected exception",
+            exc_info=True,
+        )
+    raise typer.Exit(1) from exception
+
+
 def handle_sdk_error(action: str):
-    """Convert unexpected SDK exceptions into friendly CLI errors."""
+    """Convert unexpected SDK exceptions into friendly CLI errors.
+
+    Maps SDK exception types to CLI-friendly error codes and enhances
+    error messages with CLI context. Preserves full exception context
+    including stack trace for debugging.
+    """
 
     def decorator(command_function: CommandFunction) -> CommandFunction:
         @wraps(command_function)
@@ -116,9 +447,42 @@ def wrapper(*args, **kwargs):
             try:
                 return command_function(*args, **kwargs)
             except typer.Exit:
+                # Re-raise intentional exits without modification
                 raise
+            except DashScopeException as exception:
+                # Handle known DashScope exceptions with structured error info
+                request_id = getattr(exception, "request_id", "N/A") or "N/A"
+                message = getattr(exception, "message", str(exception))
+
+                # Map exception type to CLI-friendly error code
+                exception_type_name = type(exception).__name__
+                cli_error_code = _get_cli_error_code(exception_type_name)
+                cli_error_message = _get_cli_error_message(
+                    exception_type_name,
+                    message,
+                )
+
+                err_console.print(
+                    f"[red]{action}[/red] "
+                    f"(request_id: {request_id}, code: {cli_error_code})\n"
+                    f"  {cli_error_message}",
+                    no_wrap=True,
+                )
+                # Log full traceback for debugging
+                logger.debug(
+                    f"{action} failed with DashScopeException",
+                    exc_info=True,
+                )
+                raise typer.Exit(1) from exception
             except Exception as exception:
-                error(f"{action}: {exception}")
+                # Handle unexpected exceptions with full context
+                err_console.print(f"[red]{action}:[/red] {exception}")
+                # Log full traceback for debugging unexpected errors
+                logger.debug(
+                    f"{action} failed with unexpected exception",
+                    exc_info=True,
+                )
+                raise typer.Exit(1) from exception
 
         return wrapper  # type: ignore[return-value]
 
diff --git a/dashscope/cli/deployments.py b/dashscope/cli/deployments.py
index 44f2a58..a026a70 100644
--- a/dashscope/cli/deployments.py
+++ b/dashscope/cli/deployments.py
@@ -100,7 +100,24 @@ def _print_deployments(output):
     ):
         console.print("There is no deployed model!")
         return
-    for dep in output.deployments:
+
+    # Support both dictionary and object formats
+    if isinstance(output, dict):
+        raw_deps = output.get("deployments", [])
+        from types import SimpleNamespace
+
+        deployments = [
+            SimpleNamespace(**d) if isinstance(d, dict) else d
+            for d in raw_deps
+        ]
+    else:
+        deployments = getattr(output, "deployments", [])
+
+    if not deployments:
+        console.print("There is no deployed model!")
+        return
+
+    for dep in deployments:
         console.print(
             f"Deployed_model: {dep.deployed_model}, "
             f"model: {dep.model_name}, "
diff --git a/dashscope/cli/fine_tunes.py b/dashscope/cli/fine_tunes.py
index d531752..9e28e73 100644
--- a/dashscope/cli/fine_tunes.py
+++ b/dashscope/cli/fine_tunes.py
@@ -115,11 +115,11 @@ def _stream_events(job_id: str):
         )
         return
 
-    status = (
-        rsp.output.get("status")
-        if isinstance(rsp.output, dict)
-        else getattr(rsp.output, "status", None)
-    )
+    # Support both dictionary and object formats
+    if isinstance(rsp.output, dict):
+        status = rsp.output.get("status")
+    else:
+        status = getattr(rsp.output, "status", None)
     if status in (
         TaskStatus.FAILED,
         TaskStatus.CANCELED,
@@ -252,7 +252,11 @@ def create(
     if output is None:
         error("Fine-tune creation returned empty response")
 
-    job_id = getattr(output, "job_id", None)
+    job_id = (
+        output.get("job_id")
+        if isinstance(output, dict)
+        else getattr(output, "job_id", None)
+    )
     if not job_id:
         error(
             "Fine-tune creation succeeded but missing job_id in response. "
diff --git a/dashscope/cli/generation.py b/dashscope/cli/generation.py
index b624acc..73c3617 100644
--- a/dashscope/cli/generation.py
+++ b/dashscope/cli/generation.py
@@ -7,7 +7,7 @@
 
 from dashscope.aigc import Generation
 from dashscope.cli.common import (
-    error,
+    err_console,
     handle_sdk_error,
     print_failed_message,
 )
@@ -43,7 +43,9 @@ def _build_generation_kwargs(
         try:
             kwargs["messages"] = json.loads(messages)
         except json.JSONDecodeError as exc:
-            error("--messages must be a valid JSON string")
+            err_console.print(
+                "[red]Error:[/red] --messages must be a valid JSON string",
+            )
             raise typer.Exit(1) from exc
 
     # Group simple parameters to reduce branches
@@ -178,6 +180,7 @@ def create(
                     typer.echo(usage)
             else:
                 print_failed_message(rsp)
+                break
     else:
         if response.status_code == HTTPStatus.OK:
             typer.echo(response.output)
diff --git a/dashscope/client/base_api.py b/dashscope/client/base_api.py
index fc067fc..1786500 100644
--- a/dashscope/client/base_api.py
+++ b/dashscope/client/base_api.py
@@ -1556,12 +1556,23 @@ def _handle_response(cls, response: requests.Response):
         ):
             for is_error, status_code, data in cls._handle_stream(response):
                 if is_error:
+                    try:
+                        error_data = json.loads(data)
+                        code = error_data.get("code") or error_data.get(
+                            "error_code",
+                        )
+                        message = error_data.get("message") or error_data.get(
+                            "error_message",
+                        )
+                    except json.JSONDecodeError:
+                        code = None
+                        message = data
                     yield DashScopeAPIResponse(
                         request_id=request_id,
                         status_code=status_code,
                         output=None,
-                        code="",
-                        message="",
+                        code=code,
+                        message=message,
                     )  # noqa E501
                 else:
                     yield DashScopeAPIResponse(
diff --git a/dashscope/common/error.py b/dashscope/common/error.py
index 2b49cc7..cb1cabb 100644
--- a/dashscope/common/error.py
+++ b/dashscope/common/error.py
@@ -144,3 +144,15 @@ class UploadFileException(DashScopeException):
 
 class TimeoutException(DashScopeException):
     pass
+
+
+class InvalidBaseURL(DashScopeException):
+    """Raised when the base_address is invalid.
+
+    This can happen when the scheme is missing or the URL is malformed.
+    """
+
+    class BaseUrlError:
+        """Error message pattern for invalid base URL."""
+
+        pattern = r"No scheme supplied"
diff --git a/dashscope/common/error_registry.py b/dashscope/common/error_registry.py
new file mode 100644
index 0000000..e5bc5e6
--- /dev/null
+++ b/dashscope/common/error_registry.py
@@ -0,0 +1,187 @@
+# -*- coding: utf-8 -*-
+"""Auto-generated by tools/generate.py -- DO NOT EDIT."""
+from __future__ import annotations
+from dataclasses import dataclass, field
+from typing import Dict, List
+
+__version__ = "1.0.0"
+__commit__ = "aa17fe8"
+__snapshot__ = False
+
+
+@dataclass(frozen=True)
+class PublicError:
+    key: str
+    status_code: int
+    error_code: str
+    error_msg: str
+    anthropic_error_code: str
+
+    def format_msg(self, variables: Dict[str, str] = None) -> str:
+        msg = self.error_msg
+        if variables:
+            for k, v in variables.items():
+                msg = msg.replace("{" + k + "}", v)
+        return msg
+
+
+@dataclass(frozen=True)
+class InternalErrorDef:
+    name: str
+    external: PublicError
+    message: str
+    allow_retry: bool = False
+    vars: List[str] = field(default_factory=list)
+    cause: str = ""
+    solution: str = ""
+
+    def format_message(self, variables: Dict[str, str] = None) -> str:
+        msg = self.message
+        if variables:
+            for k, v in variables.items():
+                msg = msg.replace("{" + k + "}", v)
+        return msg
+
+
+INVALID_REQUEST = PublicError(
+    key="invalid_request",
+    status_code=400,
+    error_code="BadRequestError",
+    error_msg=(
+        "The request is invalid. Please check the request and try " "again."
+    ),
+    anthropic_error_code="invalid_request_error",
+)
+
+MISSING_PARAMETER = PublicError(
+    key="missing_parameter",
+    status_code=400,
+    error_code="BadRequestError",
+    error_msg="Missing required parameter: {parameter}.",
+    anthropic_error_code="invalid_request_error",
+)
+
+CONTENT_POLICY_VIOLATION = PublicError(
+    key="content_policy_violation",
+    status_code=400,
+    error_code="BadRequestError",
+    error_msg="The request was rejected by content policy.",
+    anthropic_error_code="invalid_request_error",
+)
+
+INVALID_URL = PublicError(
+    key="invalid_url",
+    status_code=400,
+    error_code="BadRequestError",
+    error_msg="The provided URL is invalid or cannot be accessed.",
+    anthropic_error_code="invalid_request_error",
+)
+
+INVALID_FILE = PublicError(
+    key="invalid_file",
+    status_code=400,
+    error_code="BadRequestError",
+    error_msg="The provided file is invalid.",
+    anthropic_error_code="invalid_request_error",
+)
+
+AUTH_FAILED = PublicError(
+    key="auth_failed",
+    status_code=401,
+    error_code="AuthenticationError",
+    error_msg=(
+        "Authentication failed. Please provide valid "
+        "authentication credentials."
+    ),
+    anthropic_error_code="authentication_error",
+)
+
+INVALID_API_KEY = PublicError(
+    key="invalid_api_key",
+    status_code=401,
+    error_code="AuthenticationError",
+    error_msg="Incorrect API key provided.",
+    anthropic_error_code="authentication_error",
+)
+
+PERMISSION_DENIED = PublicError(
+    key="permission_denied",
+    status_code=403,
+    error_code="PermissionDeniedError",
+    error_msg="You do not have permission to access this resource.",
+    anthropic_error_code="permission_error",
+)
+
+RESOURCE_NOT_FOUND = PublicError(
+    key="resource_not_found",
+    status_code=404,
+    error_code="NotFoundError",
+    error_msg="The requested resource was not found: {resource}.",
+    anthropic_error_code="not_found_error",
+)
+
+REQUEST_TOO_LARGE = PublicError(
+    key="request_too_large",
+    status_code=413,
+    error_code="RequestTooLargeError",
+    error_msg="The request exceeds the maximum allowed size of {limit}.",
+    anthropic_error_code="request_too_large",
+)
+
+RATE_LIMIT_EXCEEDED = PublicError(
+    key="rate_limit_exceeded",
+    status_code=429,
+    error_code="RateLimitError",
+    error_msg="Rate limit exceeded. Please slow down your requests.",
+    anthropic_error_code="rate_limit_error",
+)
+
+CONCURRENCY_LIMIT_EXCEEDED = PublicError(
+    key="concurrency_limit_exceeded",
+    status_code=429,
+    error_code="RateLimitError",
+    error_msg=(
+        "Too many concurrent requests. Please reduce concurrency "
+        "and try again."
+    ),
+    anthropic_error_code="rate_limit_error",
+)
+
+INSUFFICIENT_QUOTA = PublicError(
+    key="insufficient_quota",
+    status_code=429,
+    error_code="RateLimitError",
+    error_msg=(
+        "You exceeded your current quota. Please check your plan "
+        "and billing details."
+    ),
+    anthropic_error_code="billing_error",
+)
+
+INTERNAL_ERROR = PublicError(
+    key="internal_error",
+    status_code=500,
+    error_code="InternalServerError",
+    error_msg=(
+        "The server encountered an internal error. Please try again " "later."
+    ),
+    anthropic_error_code="api_error",
+)
+
+SERVICE_UNAVAILABLE = PublicError(
+    key="service_unavailable",
+    status_code=503,
+    error_code="ServiceUnavailableError",
+    error_msg=(
+        "The service is temporarily unavailable. Please try again " "later."
+    ),
+    anthropic_error_code="overloaded_error",
+)
+
+REQUEST_TIMEOUT = PublicError(
+    key="request_timeout",
+    status_code=504,
+    error_code="GatewayTimeoutError",
+    error_msg="The request timed out. Please try again later.",
+    anthropic_error_code="timeout_error",
+)
diff --git a/dashscope/common/utils.py b/dashscope/common/utils.py
index 7a87125..bf19bba 100644
--- a/dashscope/common/utils.py
+++ b/dashscope/common/utils.py
@@ -9,7 +9,7 @@
 import threading
 from dataclasses import dataclass
 from http import HTTPStatus
-from typing import Dict
+from typing import Dict, Tuple
 from urllib.parse import urlparse
 
 import aiohttp
@@ -18,11 +18,28 @@
 from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
 from dashscope.common.api_key import get_default_api_key
 from dashscope.common.constants import SSE_CONTENT_TYPE
+from dashscope.common.error_registry import INTERNAL_ERROR
 from dashscope.common.logging import logger
 from dashscope.version import __version__
 
 
-def is_validate_fine_tune_file(file_path):
+def truncate_error_message(message: str, max_length: int = 200) -> str:
+    """Truncate error message for logging to avoid excessive log output.
+
+    Args:
+        message: The error message to truncate.
+        max_length: Maximum length of the message. Defaults to 200.
+
+    Returns:
+        Truncated message with '...' suffix if longer than max_length,
+        otherwise original message.
+    """
+    if len(message) > max_length:
+        return message[:max_length] + "..."
+    return message
+
+
+def is_validate_fine_tune_file(file_path: str) -> bool:
     with open(file_path, encoding="utf-8") as f:
         for line in f:
             try:
@@ -32,7 +49,7 @@ def is_validate_fine_tune_file(file_path):
     return True
 
 
-def _get_task_group_and_task(module_name):
+def _get_task_group_and_task(module_name: str) -> Tuple[str, str]:
     """Get task_group and task name.
     get task_group and task name based on api file __name__
 
@@ -48,41 +65,36 @@ def _get_task_group_and_task(module_name):
     return task_group, task
 
 
-def is_path(path: str):
-    """Check the input path is valid local path.
+def is_path(path: str) -> bool:
+    """Check if the input is a valid local path.
 
     Args:
-        path_or_url (str): The path.
+        path: The path to check.
 
     Returns:
-        bool: If path return True, otherwise False.
+        True if it's a valid local path, False otherwise.
     """
     url_parsed = urlparse(path)
-    if url_parsed.scheme in ("file", ""):
-        return os.path.exists(url_parsed.path)
-    else:
-        return False
+    return url_parsed.scheme in ("file", "") and os.path.exists(
+        url_parsed.path,
+    )
 
 
-def is_url(url: str):
-    """Check the input url is valid url.
+def is_url(url: str) -> bool:
+    """Check if the input is a valid URL.
 
     Args:
-        url (str): The url
+        url: The URL to check.
 
     Returns:
-        bool: If is url return True, otherwise False.
+        True if it's a valid URL, False otherwise.
     """
     url_parsed = urlparse(url)
-    # pylint: disable=simplifiable-if-statement
-    if url_parsed.scheme in ("http", "https", "oss"):
-        return True
-    else:
-        return False
+    return url_parsed.scheme in ("http", "https", "oss")
 
 
 def iter_over_async(ait):
-    loop = asyncio.get_event_loop_policy().new_event_loop()
+    loop = asyncio.new_event_loop()
     ait = ait.__aiter__()
 
     async def get_next():
@@ -106,16 +118,14 @@ def iter_thread(loop, message_queue):
                     message_queue.put((True, e, None))
                     break
         finally:
-            try:
-                loop.close()
-            except Exception:
-                pass
+            loop.close()
 
     message_queue = queue.Queue()
     x = threading.Thread(
         target=iter_thread,
         args=(loop, message_queue),
         name="iter_async_thread",
+        daemon=True,
     )
     x.start()
     while True:
@@ -123,10 +133,10 @@ def iter_thread(loop, message_queue):
         if finished:
             if error is not None:
                 yield DashScopeAPIResponse(
-                    -1,
-                    "",
-                    "Unknown",
-                    message=f"Error type: {type(error)}, message: {error}",
+                    status_code=INTERNAL_ERROR.status_code,
+                    request_id="",
+                    code=INTERNAL_ERROR.error_code,
+                    message=INTERNAL_ERROR.format_msg(),
                 )
             break
         yield obj  # pylint: disable=no-else-break
@@ -170,7 +180,7 @@ def default_headers(api_key: str = None) -> Dict[str, str]:
     return headers
 
 
-def join_url(base_url, *args):
+def join_url(base_url: str, *args: str) -> str:
     if not base_url.endswith("/"):
         base_url = base_url + "/"
     url = base_url
@@ -180,58 +190,19 @@ def join_url(base_url, *args):
     return url[:-1]
 
 
-async def _handle_aiohttp_response(response: aiohttp.ClientResponse):
-    request_id = ""
-    if response.status == HTTPStatus.OK:
-        json_content = await response.json()
-        if "request_id" in json_content:
-            request_id = json_content["request_id"]
-        return DashScopeAPIResponse(
-            request_id=request_id,
-            status_code=HTTPStatus.OK,
-            output=json_content,
-        )
-    else:
-        if "application/json" in response.content_type:
-            error = await response.json()
-            msg = ""
-            if "message" in error:
-                msg = error["message"]
-            if "request_id" in error:
-                request_id = error["request_id"]
-            return DashScopeAPIResponse(
-                request_id=request_id,
-                status_code=response.status,
-                output=None,
-                code=error["code"],
-                message=msg,
-            )
-        else:
-            msg = await response.read()
-            return DashScopeAPIResponse(
-                request_id=request_id,
-                status_code=response.status,
-                output=None,
-                code="Unknown",
-                message=msg,
-            )
-
-
 @dataclass
 class SSEEvent:
-    id: str
-    eventType: str
-    data: str
-
-    def __init__(  # pylint: disable=redefined-builtin
-        self,
-        id: str,
-        type: str,
-        data: str,
-    ):
-        self.id = id
-        self.eventType = type
-        self.data = data
+    """Server-Sent Events event representation.
+
+    Attributes:
+        id: Event ID from the 'id:' field.
+        eventType: Event type from the 'event:' field.
+        data: Event data from the 'data:' field.
+    """
+
+    id: str = ""
+    eventType: str = ""
+    data: str = ""
 
 
 def _handle_stream(response: requests.Response):
@@ -268,20 +239,48 @@ def _handle_stream(response: requests.Response):
 
 
 def _handle_error_message(error, status_code, flattened_output, headers):
-    code = None
     msg = ""
     request_id = ""
+
+    # Log original error information
+    original_code = error.get("code", error.get("error_code", ""))
+    original_message = error.get(
+        "message",
+        error.get("error_message", error.get("msg", "")),
+    )
+    logger.error(
+        "Request failed: status=%s, code=%s, message=%s",
+        status_code,
+        original_code or "unknown",
+        original_message or "unknown",
+    )
+
     if flattened_output:
         error["status_code"] = status_code
         return error
-    if "message" in error:
+
+    # Extract message, fallback to INTERNAL_ERROR.format_msg() if empty
+    if "message" in error and error["message"]:
         msg = error["message"]
-    if "msg" in error:
+    elif "msg" in error and error["msg"]:
         msg = error["msg"]
-    if "code" in error:
+    elif "error_message" in error and error["error_message"]:
+        msg = error["error_message"]
+    else:
+        msg = INTERNAL_ERROR.format_msg()
+
+    # Extract code, fallback to INTERNAL_ERROR.error_code if empty
+    if "code" in error and error["code"]:
         code = error["code"]
+    elif "error_code" in error and error["error_code"]:
+        code = error["error_code"]
+    else:
+        code = INTERNAL_ERROR.error_code
+
+    # Extract request_id
     if "request_id" in error:
         request_id = error["request_id"]
+
     return DashScopeAPIResponse(
         request_id=request_id,
         status_code=status_code,
@@ -316,22 +315,43 @@ def _handle_http_failed_response(
                     flattened_output,
                     headers,
                 )
+        # No valid error data in SSE response
+        error_message = "\n".join(msgs).strip() or INTERNAL_ERROR.format_msg()
+        error_code = INTERNAL_ERROR.error_code
+        logger.error(
+            "Request failed: status=%s, code=%s, message=%s",
+            response.status_code,
+            error_code,
+            truncate_error_message(error_message),
+        )
         return DashScopeAPIResponse(
             request_id=request_id,
             status_code=response.status_code,
-            code="Unknown",
-            message=msgs,
+            code=error_code,
+            message=error_message,
             headers=headers,
         )
     else:
         msg = response.content.decode("utf-8")
+        error_message = msg or INTERNAL_ERROR.format_msg()
+        error_code = INTERNAL_ERROR.error_code
+        logger.error(
+            "Request failed: status=%s, code=%s, message=%s",
+            response.status_code,
+            error_code,
+            truncate_error_message(error_message),
+        )
         if flattened_output:
-            return {"status_code": response.status_code, "message": msg}  # type: ignore[return-value] # pylint: disable=line-too-long # noqa: E501
+            return {  # type: ignore[return-value]
+                "status_code": response.status_code,
+                "code": error_code,
+                "message": error_message,
+            }
         return DashScopeAPIResponse(
             request_id=request_id,
             status_code=response.status_code,
-            code="Unknown",
-            message=msg,
+            code=error_code,
+            message=error_message,
             headers=headers,
         )
 
@@ -359,13 +379,33 @@ async def _handle_aio_stream(response):
 
 
 async def _handle_aiohttp_failed_response(
-    response: requests.Response,
+    response: aiohttp.ClientResponse,
     flattened_output: bool = False,
 ) -> DashScopeAPIResponse:
     request_id = ""
     headers = dict(response.headers)
     if "application/json" in response.content_type:
         error = await response.json()
+        # Pass through code, fallback to
+        # INTERNAL_ERROR.error_code if not available
+        error_code = (
+            error.get("code")
+            or error.get("error_code")
+            or INTERNAL_ERROR.error_code
+        )
+        # Pass through message, fallback to
+        # INTERNAL_ERROR.error_msg if not available
+        error_message = (
+            error.get("message")
+            or error.get("error_message")
+            or INTERNAL_ERROR.format_msg()
+        )
+        logger.error(
+            "Request failed: status=%s, code=%s, message=%s",
+            response.status,
+            error_code,
+            truncate_error_message(error_message),
+        )
         return _handle_error_message(
             error,
             response.status,
@@ -374,19 +414,34 @@ async def _handle_aiohttp_failed_response(
         )
     elif SSE_CONTENT_TYPE in response.content_type:
         error = None
+        raw_data = []
         async for _, _, data in _handle_aio_stream(response):
-            error = json.loads(data)
+            raw_data.append(data)
+            try:
+                error = json.loads(data)
+            except json.JSONDecodeError:
+                continue
         if error is None:
+            raw_content = "\n".join(raw_data).strip() if raw_data else ""
+            error_code = INTERNAL_ERROR.error_code
+            error_message = raw_content or INTERNAL_ERROR.format_msg()
+            logger.error(
+                "Request failed: status=%s, code=%s, message=%s",
+                response.status,
+                error_code,
+                truncate_error_message(error_message),
+            )
             if flattened_output:
                 return {  # type: ignore[return-value]
                     "status_code": response.status,
-                    "message": "Empty SSE error response",
+                    "code": error_code,
+                    "message": error_message,
                 }
             return DashScopeAPIResponse(
                 request_id=request_id,
                 status_code=response.status,
-                code="Unknown",
-                message="Empty SSE error response",
+                code=error_code,
+                message=error_message,
                 headers=headers,
             )
         return _handle_error_message(
@@ -397,13 +452,25 @@ async def _handle_aiohttp_failed_response(
         )
     else:
         msg = await response.text()
+        error_code = INTERNAL_ERROR.error_code
+        error_message = msg or INTERNAL_ERROR.format_msg()
+        logger.error(
+            "Request failed: status=%s, code=%s, message=%s",
+            response.status,
+            error_code,
+            truncate_error_message(error_message),
+        )
         if flattened_output:
-            return {"status_code": response.status, "message": msg}  # type: ignore[return-value] # pylint: disable=line-too-long # noqa: E501
+            return {  # type: ignore[return-value]
+                "status_code": response.status,
+                "code": error_code,
+                "message": error_message,
+            }
         return DashScopeAPIResponse(
             request_id=request_id,
             status_code=response.status,
-            code="Unknown",
-            message=msg,
+            code=error_code,
+            message=error_message,
             headers=headers,
         )
 
@@ -412,11 +479,10 @@ def _handle_http_response(
     response: requests.Response,
     flattened_output: bool = False,
 ):
-    response = _handle_http_stream_response(response, flattened_output)
-    _, output = next(response)
-    try:
-        next(response)
-    except StopIteration:
+    response_gen = _handle_http_stream_response(response, flattened_output)
+    _, output = next(response_gen)
+    # Consume remaining items to ensure generator completes
+    for _ in response_gen:
         pass
     return output
 
@@ -457,19 +523,28 @@ def _handle_http_stream_response(
                             usage=usage,
                             headers=headers,
                         )
-                except json.JSONDecodeError as e:
+                except json.JSONDecodeError:
+                    error_code = INTERNAL_ERROR.error_code
+                    error_message = event.data or INTERNAL_ERROR.format_msg()
+                    logger.error(
+                        "Request failed: status=%s, code=%s, message=%s",
+                        response.status_code,
+                        error_code,
+                        truncate_error_message(error_message),
+                    )
                     if flattened_output:
                         yield event.eventType, {
                             "status_code": response.status_code,
-                            "message": e.message,
+                            "code": error_code,
+                            "message": error_message,
                         }
                     else:
                         yield event.eventType, DashScopeAPIResponse(
                             request_id=request_id,
-                            status_code=HTTPStatus.BAD_REQUEST,
+                            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                             output=None,
-                            code="Unknown",
-                            message=event.data,
+                            code=error_code,
+                            message=error_message,
                             headers=headers,
                         )
                     continue
@@ -480,17 +555,39 @@ def _handle_http_stream_response(
                         "message": event.data,
                     }
                 else:
-                    msg = json.loads(event.data)
-                    yield event.eventType, DashScopeAPIResponse(
-                        request_id=request_id,
-                        status_code=status_code,
-                        output=None,
-                        code=msg["code"]
-                        if "code" in msg
-                        else None,  # noqa E501
-                        message=msg["message"] if "message" in msg else None,
-                        headers=headers,
-                    )  # noqa E501
+                    try:
+                        msg = json.loads(event.data)
+                        yield event.eventType, DashScopeAPIResponse(
+                            request_id=request_id,
+                            status_code=status_code,
+                            output=None,
+                            code=msg.get("code")
+                            or msg.get("error_code")
+                            or f"http_{status_code}",
+                            message=msg.get("message")
+                            or msg.get("error_message")
+                            or f"HTTP {status_code} error",
+                            headers=headers,
+                        )  # noqa E501
+                    except json.JSONDecodeError:
+                        error_code = INTERNAL_ERROR.error_code
+                        error_message = (
+                            event.data or INTERNAL_ERROR.format_msg()
+                        )
+                        logger.error(
+                            "Request failed: status=%s, code=%s, message=%s",
+                            status_code,
+                            error_code,
+                            truncate_error_message(error_message),
+                        )
+                        yield event.eventType, DashScopeAPIResponse(
+                            request_id=request_id,
+                            status_code=status_code,
+                            output=None,
+                            code=error_code,
+                            message=error_message,
+                            headers=headers,
+                        )
     # pylint: disable=consider-using-in
     elif (
         response.status_code == HTTPStatus.OK
diff --git a/dashscope/finetune/agentic_rl.py b/dashscope/finetune/agentic_rl.py
index 61b9c18..93832b4 100644
--- a/dashscope/finetune/agentic_rl.py
+++ b/dashscope/finetune/agentic_rl.py
@@ -3,7 +3,7 @@
 
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
-from typing import Union, List, Optional, ClassVar, Dict, Any
+from typing import Union, List, Optional, ClassVar, Dict, Any, Tuple
 from typing_extensions import Self
 
 from dashscope.client.base_api import CreateMixin
@@ -81,7 +81,7 @@ async def register_functions(
             ]
         ] = None,
         lazy_load: Optional[bool] = True,
-    ) -> tuple[
+    ) -> Tuple[
         List[str],
         List[str],
         List[str],
@@ -127,7 +127,7 @@ async def upload_datasets(
         datasets: Optional[List[Dataset]] = None,
         training_files: Optional[Union[List[str], str]] = None,
         validation_files: Optional[Union[List[str], str]] = None,
-    ) -> tuple[List[str], List[str]]:
+    ) -> Tuple[List[str], List[str]]:
         if datasets:
             self.tuning.datasets = datasets
 
diff --git a/dashscope/finetune/reinforcement/common/model.py b/dashscope/finetune/reinforcement/common/model.py
index 8f4123b..879c277 100644
--- a/dashscope/finetune/reinforcement/common/model.py
+++ b/dashscope/finetune/reinforcement/common/model.py
@@ -1171,7 +1171,7 @@ class TuningModel(Models, BaseModel):
     async def register_functions(
         self,
         lazy_load: Optional[bool] = True,
-    ) -> tuple[
+    ) -> Tuple[
         List[str],
         List[str],
         List[str],
@@ -1265,7 +1265,7 @@ async def upload_datasets(
         self,
         training_files: Union[List[str], str] = None,
         validation_files: Union[List[str], str] = None,
-    ) -> tuple[List[str], List[str]]:
+    ) -> Tuple[List[str], List[str]]:
         """Register and validate training/validation datasets."""
         uploaded_training_ids = []
         uploaded_validation_ids = []
diff --git a/tests/unit/mock_request_base.py b/tests/unit/mock_request_base.py
index c5c016b..ef3ee37 100644
--- a/tests/unit/mock_request_base.py
+++ b/tests/unit/mock_request_base.py
@@ -20,7 +20,7 @@ def setup_class(cls):
         super().setup_class()
         dashscope.base_http_api_url = "http://localhost:8089/api/v1/"
         dashscope.base_websocket_api_url = (
-            "http://localhost:8089/api-ws/v1/inference"
+            "ws://localhost:8089/api-ws/v1/inference"
         )
         dashscope.api_key = "default"
         dashscope.api_protocol = "http"
diff --git a/tests/unit/test_cli_common.py b/tests/unit/test_cli_common.py
new file mode 100644
index 0000000..37e5a97
--- /dev/null
+++ b/tests/unit/test_cli_common.py
@@ -0,0 +1,263 @@
+# -*- coding: utf-8 -*-
+"""Test cases for dashscope/cli/common.py error handling improvements."""
+from http import HTTPStatus
+from unittest.mock import Mock
+
+import pytest
+import typer
+
+from dashscope.cli.common import (
+    print_failed_message,
+    ensure_ok,
+    handle_sdk_error,
+)
+from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
+from dashscope.common.error import AuthenticationError
+
+
+class TestPrintFailedMessage:
+    """Test print_failed_message with various response scenarios."""
+
+    def test_complete_response(self, capsys):
+        """Test with all fields present."""
+        rsp = DashScopeAPIResponse(
+            status_code=500,
+            request_id="req_123",
+            code="ServerError",
+            message="Internal server error",
+        )
+        print_failed_message(rsp)
+        captured = capsys.readouterr()
+        assert "req_123" in captured.err
+        assert "500" in captured.err
+        assert "ServerError" in captured.err
+        assert "Internal server error" in captured.err
+
+    def test_missing_request_id(self, capsys):
+        """Test when request_id is missing."""
+        rsp = Mock()
+        rsp.status_code = 400
+        rsp.code = "BadRequest"
+        rsp.message = "Invalid parameter"
+        # Simulate missing request_id attribute
+        del rsp.request_id
+
+        print_failed_message(rsp)
+        captured = capsys.readouterr()
+        # Missing request_id should not be displayed (empty fields are omitted)
+        assert "request_id:" not in captured.err
+        # Error code is kept in original camelCase format
+        assert "BadRequest" in captured.err
+        assert "400" in captured.err
+
+    def test_empty_code_and_message(self, capsys):
+        """Test when code and message are empty strings."""
+        rsp = DashScopeAPIResponse(
+            status_code=503,
+            request_id="req_456",
+            code="",
+            message="",
+        )
+        print_failed_message(rsp)
+        captured = capsys.readouterr()
+        # Should not show empty code/message fields
+        # Use word boundary check: "code: " with space after colon
+        assert ", code: " not in captured.err
+        assert ", message: " not in captured.err
+        assert "req_456" in captured.err
+
+    def test_none_attributes(self, capsys):
+        """Test when attributes are None."""
+        rsp = Mock()
+        rsp.status_code = 502
+        rsp.request_id = None
+        rsp.code = None
+        rsp.message = None
+
+        print_failed_message(rsp)
+        captured = capsys.readouterr()
+        # None values should not be displayed
+        assert ", request_id:" not in captured.err
+        assert ", code: " not in captured.err
+        assert ", message: " not in captured.err
+        assert "502" in captured.err
+
+
+class TestEnsureOk:
+    """Test ensure_ok with various response scenarios."""
+
+    def test_successful_response(self):
+        """Test with successful HTTP 200 and no business error."""
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            request_id="req_ok",
+            code="",
+            message="",
+            output={"result": "success"},
+        )
+        result = ensure_ok(rsp)
+        assert result == {"result": "success"}
+
+    def test_http_error(self, capsys):
+        """Test with HTTP error status."""
+        rsp = DashScopeAPIResponse(
+            status_code=404,
+            request_id="req_404",
+            code="NotFound",
+            message="Resource not found",
+        )
+
+        with pytest.raises(typer.Exit):
+            ensure_ok(rsp)
+
+        captured = capsys.readouterr()
+        # Should only print once (not duplicated)
+        # Error message now uses "Request failed" instead of "Failed"
+        assert captured.err.count("Request failed") == 1
+
+    def test_business_error_in_dict_output(self, capsys):
+        """Test with HTTP 200 but business error in dict output."""
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            request_id="req_biz_err",
+            code="",
+            message="",
+            output={"code": "InvalidParameter", "message": "Model not found"},
+        )
+
+        with pytest.raises(typer.Exit):
+            ensure_ok(rsp)
+
+        captured = capsys.readouterr()
+        # Rich may wrap long lines and add extra spaces, normalize whitespace
+        normalized_err = " ".join(captured.err.split())
+        # Error code is kept in original camelCase format
+        assert "InvalidParameter" in normalized_err
+        assert "Model not found" in normalized_err
+
+    def test_business_error_without_message(self, capsys):
+        """Test business error without message field."""
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            request_id="req_no_msg",
+            output={"code": "SomeError"},  # No message field
+        )
+
+        with pytest.raises(typer.Exit):
+            ensure_ok(rsp)
+
+        captured = capsys.readouterr()
+        # Should show improved fallback message
+        # (normalize all whitespace for Rich formatting)
+        normalized_err = " ".join(captured.err.split())
+        assert "API returned error code without message" in normalized_err
+
+    def test_none_output(self, capsys):
+        """Test when output is None despite HTTP 200."""
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            request_id="req_null",
+            output=None,
+        )
+
+        with pytest.raises(typer.Exit):
+            ensure_ok(rsp)
+
+        captured = capsys.readouterr()
+        assert "no output data" in captured.err
+
+    def test_skip_business_error_check(self):
+        """Test with check_business_error=False."""
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            output={
+                "code": "AsyncTaskPending",
+                "message": "Task is processing",
+            },
+        )
+
+        # Should not raise even though there's a code in output
+        result = ensure_ok(rsp, check_business_error=False)
+        assert result == {
+            "code": "AsyncTaskPending",
+            "message": "Task is processing",
+        }
+
+    def test_object_output_with_error(self, capsys):
+        """Test with object output containing error fields."""
+        mock_output = Mock()
+        mock_output.code = "ObjectError"
+        mock_output.message = "Object-level error"
+
+        rsp = DashScopeAPIResponse(
+            status_code=HTTPStatus.OK,
+            request_id="req_obj",
+            output=mock_output,
+        )
+
+        with pytest.raises(typer.Exit):
+            ensure_ok(rsp)
+
+        captured = capsys.readouterr()
+        assert "ObjectError" in captured.err
+
+
+class TestHandleSdkError:
+    """Test handle_sdk_error decorator."""
+
+    def test_dashscope_exception_handling(self, capsys):
+        """Test handling of DashScopeException."""
+
+        @handle_sdk_error("Test action")
+        def failing_function():
+            # Create exception properly using __init__ with positional args
+            exc = AuthenticationError()
+            exc.request_id = "req_auth"
+            exc.code = "AuthFailed"
+            exc.message = "Invalid API key"
+            raise exc
+
+        with pytest.raises(typer.Exit):
+            failing_function()
+
+        captured = capsys.readouterr()
+        assert "Test action" in captured.err
+        assert "req_auth" in captured.err
+        # Error code is kept in original camelCase format (exception type name)
+        assert "AuthenticationError" in captured.err
+
+    def test_generic_exception_handling(self, capsys):
+        """Test handling of generic exceptions."""
+
+        @handle_sdk_error("Generic test")
+        def generic_failing_function():
+            raise ValueError("Something went wrong")
+
+        with pytest.raises(typer.Exit):
+            generic_failing_function()
+
+        captured = capsys.readouterr()
+        assert "Generic test" in captured.err
+        assert "Something went wrong" in captured.err
+
+    def test_typer_exit_passthrough(self):
+        """Test that typer.Exit is re-raised without modification."""
+
+        @handle_sdk_error("Should not catch this")
+        def intentional_exit():
+            raise typer.Exit(code=2)
+
+        with pytest.raises(typer.Exit) as exc_info:
+            intentional_exit()
+
+        assert exc_info.value.exit_code == 2
+
+    def test_successful_function_passthrough(self):
+        """Test that successful functions work normally."""
+
+        @handle_sdk_error("Success test")
+        def success_function():
+            return "success"
+
+        result = success_function()
+        assert result == "success"
diff --git a/tests/unit/test_cli_main.py b/tests/unit/test_cli_main.py
index d4c839c..4b178ef 100644
--- a/tests/unit/test_cli_main.py
+++ b/tests/unit/test_cli_main.py
@@ -46,7 +46,7 @@ def mock_list(**kwargs):
         combined_output = captured_output.out + captured_output.err
 
         assert exception_info.value.code == 1
-        assert "No api key provided." in combined_output
+        assert "No api key provided" in combined_output
         assert "Traceback" not in combined_output
 
     def test_top_level_help_shows_global_api_key(self):
diff --git a/tests/unit/test_utf8_encoding.py b/tests/unit/test_utf8_encoding.py
index ebee93c..965f719 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; charset=utf-8"}
 
     async def json(self):
         return {"output": {"text": "ok"}, "request_id": "test-123"}