refactor: modernize TLS API, deduplicate validation, and improve error context

Author: Fankouzu

src/a2a/client/tls.py

@@ -54,13 +54,15 @@ def create_ssl_context(self) -> ssl.SSLContext:
         if isinstance(self.verify, ssl.SSLContext):
             return self.verify
 
-        protocol_map = {
-            'TLSv1_2': ssl.PROTOCOL_TLSv1_2,
-            'TLSv1_3': ssl.PROTOCOL_TLS_CLIENT,
-        }
-        protocol = protocol_map.get(self.min_version, ssl.PROTOCOL_TLS_CLIENT)
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
 
-        context = ssl.SSLContext(protocol)
+        version_map = {
+            'TLSv1_2': ssl.TLSVersion.TLSv1_2,
+            'TLSv1_3': ssl.TLSVersion.TLSv1_3,
+        }
+        context.minimum_version = version_map.get(
+            self.min_version, ssl.TLSVersion.TLSv1_2
+        )
 
         if self.ca_cert:
             context.load_verify_locations(cafile=str(self.ca_cert))
@@ -220,15 +222,17 @@ def create_server_ssl_context(
     Returns:
         Configured ssl.SSLContext for server use.
     """
-    protocol_map = {
-        'TLSv1_2': ssl.PROTOCOL_TLSv1_2,
-        'TLSv1_3': ssl.PROTOCOL_TLS_SERVER,
-    }
-    protocol = protocol_map.get(min_version, ssl.PROTOCOL_TLS_SERVER)
-
-    context = ssl.SSLContext(protocol)
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
     context.load_cert_chain(str(cert_file), str(key_file))
 
+    version_map = {
+        'TLSv1_2': ssl.TLSVersion.TLSv1_2,
+        'TLSv1_3': ssl.TLSVersion.TLSv1_3,
+    }
+    context.minimum_version = version_map.get(
+        min_version, ssl.TLSVersion.TLSv1_2
+    )
+
     if ca_cert:
         context.load_verify_locations(cafile=str(ca_cert))
 

src/a2a/validation.py

@@ -227,6 +227,52 @@ def validate_message(
         ) from e
 
 
+def _validate_against_types(
+    data: dict[str, Any],
+    model_types: tuple[type[BaseModel], ...],
+    category_name: str,
+) -> BaseModel:
+    """Validate data against multiple model types and return first match.
+
+    Args:
+        data: Raw data to validate.
+        model_types: Tuple of model types to try.
+        category_name: Name of the category for error messages (e.g., 'request', 'response').
+
+    Returns:
+        The validated model instance.
+
+    Raises:
+        ValidationError: If data doesn't match any of the provided types.
+    """
+    errors: list[dict[str, Any]] = []
+
+    for model_type in model_types:
+        try:
+            return validate_message(data, model_type, strict=False)
+        except ValidationError as e:
+            errors.append(
+                {
+                    'type': model_type.__name__,
+                    'path': e.errors[0].get('path', []) if e.errors else [],
+                    'message': e.errors[0].get('message', str(e))
+                    if e.errors
+                    else str(e),
+                }
+            )
+
+    error_details = '; '.join(
+        f'{e["type"]}: {e["message"]} (path: {".".join(map(str, e["path"])) or "root"})'
+        for e in errors
+    )
+    raise ValidationError(
+        f'Data does not match any known A2A {category_name} type. '
+        f'Attempted types: {[e["type"] for e in errors]}. Details: {error_details}',
+        errors=errors,
+        instance=data,
+    )
+
+
 def validate_request(data: dict[str, Any]) -> BaseModel:
     """Validate and parse an A2A request message.
 
@@ -252,33 +298,7 @@ def validate_request(data: dict[str, Any]) -> BaseModel:
         TaskResubscriptionRequest,
     )
 
-    errors: list[dict[str, Any]] = []
-
-    for model_type in request_types:
-        try:
-            return validate_message(data, model_type, strict=False)
-        except ValidationError:
-            continue
-
-    raise ValidationError(
-        'Data does not match any known A2A request type',
-        errors=errors,
-        instance=data,
-    )
-
-    errors: list[dict[str, Any]] = []
-
-    for model_type in request_types:
-        try:
-            return validate_message(data, model_type, strict=False)
-        except ValidationError:
-            continue
-
-    raise ValidationError(
-        'Data does not match any known A2A request type',
-        errors=errors,
-        instance=data,
-    )
+    return _validate_against_types(data, request_types, 'request')
 
 
 def validate_response(data: dict[str, Any]) -> BaseModel:
@@ -302,27 +322,7 @@ def validate_response(data: dict[str, Any]) -> BaseModel:
         GetTaskPushNotificationConfigResponse,
     )
 
-    for model_type in response_types:
-        try:
-            return validate_message(data, model_type, strict=False)
-        except ValidationError:
-            continue
-
-    raise ValidationError(
-        'Data does not match any known A2A response type',
-        instance=data,
-    )
-
-    for model_type in response_types:
-        try:
-            return validate_message(data, model_type, strict=False)
-        except ValidationError:
-            continue
-
-    raise ValidationError(
-        'Data does not match any known A2A response type',
-        instance=data,
-    )
+    return _validate_against_types(data, response_types, 'response')
 
 
 class MessageValidator: