smithy-lang · jonathan343 · Jun 12, 2026 · Jun 18, 2026 · Jul 14, 2026
@@ -653,6 +653,7 @@ class $4L:
                     ""\"An asynchronous HTTP client solely for testing purposes.""\"
 
                     TIMEOUT_EXCEPTIONS = ()
+                    SUPPORTS_DUPLEX_STREAMING: bool = True
 
                     def __init__(self, *, client_config: $5T | None = None):
                         self._client_config = client_config
@@ -668,6 +669,7 @@ class $8L:
                     ""\"An asynchronous HTTP client solely for testing purposes.""\"
 
                     TIMEOUT_EXCEPTIONS = ()
+                    SUPPORTS_DUPLEX_STREAMING: bool = True
 
                     def __init__(
                         self,

@@ -145,7 +145,11 @@ private static List<ConfigProperty> getProtocolProperties(GenerationContext cont
                                 .namespace("smithy_core.aio.interfaces", ".")
                                 .build())
                         .build())
-                .documentation("The transport to use to send requests (e.g. an HTTP client).");
+                .documentation("The transport to use to send requests (e.g. an HTTP client). "
+                        + "Operations with bidirectional event streams require a "
+                        + "DuplexClientTransport, such as AWSCRTHTTPClient. Transports are "
+                        + "assumed not to support duplex streaming unless they explicitly set "
+                        + "SUPPORTS_DUPLEX_STREAMING to True.");
 
         if (context.applicationProtocol().isHttpProtocol()) {
             properties.addAll(HTTP_PROPERTIES);

@@ -258,6 +258,15 @@ which takes a request and some configuration and asynchronously return a respons
 Having a minimal interface makes it much easier to implement these interfaces on top of
 a variety http libraries.
 
+Clients that support duplex (bidirectional) event streaming, which in practice
+requires HTTP/2, implement the `DuplexClientTransport` protocol by setting the
+`SUPPORTS_DUPLEX_STREAMING` class attribute to `True`. Clients are assumed not to
+support it otherwise, and duplex stream operations invoked with such a client fail
+fast with an `UnsupportedTransportError`. This prevents clients that cannot support
+duplex streaming, such as `AIOHTTPClient`, from attempting these operations. This
+capability declaration does not guarantee that a particular connection negotiates
+HTTP/2.
+
 ```python
 @dataclass(kw_only=True)
 class HTTPRequestConfiguration:

@@ -0,0 +1,4 @@
+{
+  "type": "enhancement",
+  "description": "Added `DuplexClientTransport` so transports can declare support for duplex (bidirectional) event streaming. `RequestPipeline.duplex_stream` now fails fast with an `UnsupportedTransportError` when the configured transport does not declare support."
+}
@@ -12,7 +12,12 @@
 from ..auth import AuthParams
 from ..deserializers import DeserializeableShape, ShapeDeserializer
 from ..endpoints import EndpointResolverParams
-from ..exceptions import ClientTimeoutError, RetryError, SmithyError
+from ..exceptions import (
+    ClientTimeoutError,
+    RetryError,
+    SmithyError,
+    UnsupportedTransportError,
+)
 from ..interceptors import (
     InputContext,
     Interceptor,
@@ -197,6 +202,18 @@ async def duplex_stream[
         :param output_event_type: The event type to receive in the output stream.
         :param event_deserializer: The method used to deserialize events.
         """
+        # The transport is assumed not to support duplex streaming unless it
+        # explicitly declares otherwise.
+        if getattr(self.transport, "SUPPORTS_DUPLEX_STREAMING", False) is not True:
+            raise UnsupportedTransportError(
+                f"The configured transport ({type(self.transport).__name__}) does "
+                f"not support duplex (bidirectional) event streaming, which is "
+                f"required by the {call.operation.schema.id} operation. Use a "
+                f"transport that does, such as "
+                f"smithy_http.aio.crt.AWSCRTHTTPClient. Custom transports that "
+                f"support duplex streaming can implement DuplexClientTransport "
+                f"by setting SUPPORTS_DUPLEX_STREAMING to True."
+            )
         request_future = Future[RequestContext[I, TRequest]]()
         execute_task = asyncio.create_task(self._execute_request(call, request_future))
         request_context = await request_future

@@ -1,7 +1,7 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
 from collections.abc import AsyncIterable, Callable
-from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Any, Literal, Protocol, runtime_checkable
 
 from ...documents import TypeRegistry
 from ...endpoints import EndpointResolverParams
@@ -86,7 +86,11 @@ async def resolve_endpoint(self, params: EndpointResolverParams[Any]) -> Endpoin
 
 
 class ClientTransport[I: Request, O: Response](Protocol):
-    """Protocol-agnostic representation of a client transport (e.g. an HTTP client)."""
+    """Protocol-agnostic representation of a client transport (e.g. an HTTP client).
+
+    Transports that support duplex (bidirectional) event streaming can implement
+    :py:class:`DuplexClientTransport`.
+    """
 
     TIMEOUT_EXCEPTIONS: tuple[type[Exception], ...]
 
@@ -95,6 +99,16 @@ async def send(self, request: I) -> O:
         ...
 
 
+class DuplexClientTransport[I: Request, O: Response](ClientTransport[I, O], Protocol):
+    """A client transport that supports duplex (bidirectional) event streaming.
+
+    Duplex transports can read response data while the request body is still being
+    written. HTTP transports typically require HTTP/2 for this capability.
+    """
+
+    SUPPORTS_DUPLEX_STREAMING: Literal[True]
+
+
 class ClientProtocol[I: Request, O: Response](Protocol):
     """A protocol used by a client to communicate with a server."""
 

@@ -114,5 +114,10 @@ class UnsupportedStreamError(SmithyError):
     streams are not supported."""
 
 
+class UnsupportedTransportError(SmithyError):
+    """Indicates that an operation requires a transport capability that the configured
+    transport does not declare support for (e.g. duplex event streaming)."""
+
+
 class EndpointResolutionError(SmithyError):
     """Exception type for all exceptions raised by endpoint resolution."""
@@ -0,0 +1,207 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from dataclasses import dataclass
+from typing import Any, Self, cast
+
+from smithy_core import URI
+from smithy_core.aio.client import ClientCall, RequestPipeline
+from smithy_core.aio.interfaces import ClientProtocol
+from smithy_core.deserializers import ShapeDeserializer
+from smithy_core.documents import TypeRegistry
+from smithy_core.endpoints import EndpointResolverParams
+from smithy_core.interceptors import InterceptorChain
+from smithy_core.schemas import APIOperation, Schema
+from smithy_core.serializers import ShapeSerializer
+from smithy_core.shapes import ShapeID, ShapeType
+from smithy_core.traits import StreamingTrait
+from smithy_core.types import TypedProperties
+
+_STRING = Schema(id=ShapeID("smithy.api#String"), shape_type=ShapeType.STRING)
+
+_EVENTS = Schema.collection(
+    id=ShapeID("com.example#Events"),
+    shape_type=ShapeType.UNION,
+    members={"message": {"target": _STRING}},
+)
+
+_INPUT_SCHEMA = Schema.collection(
+    id=ShapeID("com.example#StreamingInput"),
+    members={"events": {"target": _EVENTS, "traits": [StreamingTrait()]}},
+)
+
+_OUTPUT_SCHEMA = Schema.collection(
+    id=ShapeID("com.example#StreamingOutput"),
+    members={"events": {"target": _EVENTS, "traits": [StreamingTrait()]}},
+)
+
+
+class StubInput:
+    def serialize(self, serializer: ShapeSerializer) -> None:
+        pass
+
+
+class StubOutput:
+    @classmethod
+    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
+        return cls()
+
+
+class StubEvent:
+    def serialize(self, serializer: ShapeSerializer) -> None:
+        pass
+
+    @classmethod
+    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
+        return cls()
+
+
+OPERATION = APIOperation(
+    input=StubInput,
+    output=StubOutput,
+    schema=Schema(
+        id=ShapeID("com.example#StreamingOperation"),
+        shape_type=ShapeType.OPERATION,
+    ),
+    input_schema=_INPUT_SCHEMA,
+    output_schema=_OUTPUT_SCHEMA,
+    error_registry=TypeRegistry({}),
+    effective_auth_schemes=[],
+    error_schemas=[],
+)
+
+
+class StubRequest:
+    def __init__(self) -> None:
+        self.destination = URI(host="example.com")
+        self.body = b""
+
+    async def consume_body_async(self) -> bytes:
+        return b""
+
+    def consume_body(self) -> bytes:
+        return b""
+
+
+class StubResponse:
+    body = b""
+
+    async def consume_body_async(self) -> bytes:
+        return b""
+
+    def consume_body(self) -> bytes:
+        return b""
+
+
+class StubEventPublisher:
+    async def send(self, event: Any) -> None:
+        pass
+
+    async def close(self) -> None:
+        pass
+
+
+class StubEventReceiver:
+    async def receive(self) -> Any:
+        return None
+
+    async def close(self) -> None:
+        pass
+
+
+class StubProtocol:
+    def __init__(self) -> None:
+        self.serialize_request_calls = 0
+        self.set_service_endpoint_calls = 0
+        self.deserialize_response_calls = 0
+        self.create_event_publisher_calls = 0
+        self.create_event_receiver_calls = 0
+
+    @property
+    def id(self) -> ShapeID:
+        return ShapeID("com.example#testProtocol")
+
+    def serialize_request(self, **kwargs: Any) -> StubRequest:
+        self.serialize_request_calls += 1
+        return StubRequest()
+
+    def set_service_endpoint(self, *, request: Any, endpoint: Any) -> Any:
+        self.set_service_endpoint_calls += 1
+        return request
+
+    async def deserialize_response(self, **kwargs: Any) -> StubOutput:
+        self.deserialize_response_calls += 1
+        return StubOutput()
+
+    def create_event_publisher(self, **kwargs: Any) -> StubEventPublisher:
+        self.create_event_publisher_calls += 1
+        return StubEventPublisher()
+
+    def create_event_receiver(self, **kwargs: Any) -> StubEventReceiver:
+        self.create_event_receiver_calls += 1
+        return StubEventReceiver()
+
+
+class StubEndpoint:
+    def __init__(self) -> None:
+        self.uri = URI(host="example.com")
+        self.properties = TypedProperties()
+
+
+class StubEndpointResolver:
+    async def resolve_endpoint(self, params: EndpointResolverParams[Any]) -> Any:
+        return StubEndpoint()
+
+
+class StubAuthResolver:
+    def resolve_auth_scheme(self, *, auth_parameters: Any) -> list[Any]:
+        return []
+
+
+class UndeclaredTransport:
+    """A transport that does not declare whether it supports duplex streaming."""
+
+    TIMEOUT_EXCEPTIONS: tuple[type[Exception], ...] = ()
+
+    def __init__(self) -> None:
+        self.send_calls = 0
+
+    async def send(self, request: Any) -> StubResponse:
+        self.send_calls += 1
+        return StubResponse()
+
+
+class NonDuplexTransport(UndeclaredTransport):
+    SUPPORTS_DUPLEX_STREAMING = False
+
+
+class DuplexTransport(UndeclaredTransport):
+    SUPPORTS_DUPLEX_STREAMING = True
+
+
+@dataclass
+class PipelineHarness:
+    protocol: StubProtocol
+    transport: UndeclaredTransport
+    pipeline: RequestPipeline[Any, Any]
+
+
+def pipeline_harness(transport: UndeclaredTransport) -> PipelineHarness:
+    protocol = StubProtocol()
+    pipeline = RequestPipeline(
+        protocol=cast("ClientProtocol[Any, Any]", protocol),
+        transport=transport,
+    )
+    return PipelineHarness(protocol=protocol, transport=transport, pipeline=pipeline)
+
+
+def client_call() -> ClientCall[Any, Any]:
+    return ClientCall(
+        input=StubInput(),
+        operation=OPERATION,
+        context=TypedProperties(),
+        interceptor=InterceptorChain([]),
+        auth_scheme_resolver=StubAuthResolver(),
+        supported_auth_schemes={},
+        endpoint_resolver=StubEndpointResolver(),
+        retry_strategy=None,  # type: ignore[arg-type] # unused for streaming input
+    )