feat: add support for ResponseTransferMode on Api resource (#3881)

bchampp · reedham-aws · web-flow · commit cf87113fe8db · 2026-03-03T10:40:04.000-08:00
Co-authored-by: Reed Hamilton &lt;reedham@amazon.com&gt;
diff --git a/samtranslator/internal/schema_source/aws_serverless_function.py b/samtranslator/internal/schema_source/aws_serverless_function.py
@@ -308,6 +308,7 @@ class ApiEventProperties(BaseModel):
         "TimeoutInMillis",
         ["AWS::ApiGateway::Method.Integration", "TimeoutInMillis"],
     )
+    ResponseTransferMode: Optional[PassThroughProp] = apieventproperties("ResponseTransferMode")
 
 
 class ApiEvent(BaseModel):
diff --git a/samtranslator/internal/schema_source/sam-docs.json b/samtranslator/internal/schema_source/sam-docs.json
@@ -150,7 +150,8 @@
       "RequestModel": "Request model to use for this specific Api\\$1Path\\$1Method. This should reference the name of a model specified in the `Models` section of an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource.  \n*Type*: [RequestModel](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestmodel.html)  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
       "RequestParameters": "Request parameters configuration for this specific Api\\$1Path\\$1Method. All parameter names must start with `method.request` and must be limited to `method.request.header`, `method.request.querystring`, or `method.request.path`.  \nA list can contain both parameter name strings and [RequestParameter](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestparameter.html) objects. For strings, the `Required` and `Caching` properties will default to `false`.  \n*Type*: List of [ String \\$1 [RequestParameter](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestparameter.html) ]  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
       "RestApiId": "Identifier of a RestApi resource, which must contain an operation with the given path and method. Typically, this is set to reference an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource defined in this template.  \nIf you don't define this property, AWS SAM creates a default [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource using a generated `OpenApi` document. That resource contains a union of all paths and methods defined by `Api` events in the same template that do not specify a `RestApiId`.  \nThis cannot reference an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource defined in another template.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
-      "TimeoutInMillis": "Custom timeout between 50 and 29,000 milliseconds.  \nWhen you specify this property, AWS SAM modifies your OpenAPI definition. The OpenAPI definition must be specified inline using the `DefinitionBody` property. \n*Type*: Integer  \n*Required*: No  \n*Default*: 29,000 milliseconds or 29 seconds  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent."
+      "TimeoutInMillis": "Custom timeout between 50 and 29,000 milliseconds.  \nWhen you specify this property, AWS SAM modifies your OpenAPI definition. The OpenAPI definition must be specified inline using the `DefinitionBody` property. \n*Type*: Integer  \n*Required*: No  \n*Default*: 29,000 milliseconds or 29 seconds  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
+      "ResponseTransferMode": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*CloudFormation compatibility*: This property is passed directly to the [`ResponseTransferMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-apigateway-method-integration.html#cfn-apigateway-method-integration-responsetransfermode) property of an `AWS::ApiGateway::Method Integration`\\."
     },
     "sam-property-function-apifunctionauth": {
       "ApiKeyRequired": "Requires an API key for this API, path, and method.  \n*Type*: Boolean  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",
@@ -886,4 +887,4 @@
       "UseAliasAsEventTarget": "Indicate whether or not to pass the alias, created by using the `AutoPublishAlias` property, to the events source's target defined with [Events](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/#sam-statemachine-events.html#sam-statemachine-events).  \nSpecify `True` to use the alias as the events' target.  \n*Type*: Boolean  \n*Required*: No  \n*Default*: `False`  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent."
     }
   }
-}
+}
diff --git a/samtranslator/model/eventsources/push.py b/samtranslator/model/eventsources/push.py
@@ -689,6 +689,7 @@ class Api(PushEventSource):
         "RequestModel": PropertyType(False, IS_DICT),
         "RequestParameters": PropertyType(False, IS_LIST),
         "TimeoutInMillis": PropertyType(False, IS_INT),
+        "ResponseTransferMode": PropertyType(False, IS_STR),
     }
 
     Path: str
@@ -699,6 +700,7 @@ class Api(PushEventSource):
     RequestModel: Optional[Dict[str, Any]]
     RequestParameters: Optional[List[Any]]
     TimeoutInMillis: Optional[PassThrough]
+    ResponseTransferMode: Optional[PassThrough]
 
     def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]:
         """
@@ -863,7 +865,7 @@ def _add_swagger_integration(  # type: ignore[no-untyped-def] # noqa: PLR0912, P
             swagger_body = SwaggerEditor.gen_skeleton()
 
         partition = ArnGenerator.get_partition_name()
-        uri = _build_apigw_integration_uri(function, partition)  # type: ignore[no-untyped-call]
+        uri = _build_apigw_integration_uri(function, partition, self.ResponseTransferMode)  # type: ignore[no-untyped-call]
 
         editor = SwaggerEditor(swagger_body)
 
@@ -882,7 +884,15 @@ def _add_swagger_integration(  # type: ignore[no-untyped-def] # noqa: PLR0912, P
         sam_expect(method_auth, self.relative_id, "Auth", is_sam_event=True).to_be_a_map()
         api_auth = api.get("Auth") or Py27Dict()
         sam_expect(api_auth, api_id, "Auth").to_be_a_map()
-        editor.add_lambda_integration(self.Path, self.Method, uri, method_auth, api_auth, condition=condition)
+        editor.add_lambda_integration(
+            self.Path,
+            self.Method,
+            uri,
+            method_auth,
+            api_auth,
+            condition=condition,
+            invoke_mode=self.ResponseTransferMode,
+        )
 
         # self.Stage is not None as it is set in _get_permissions()
         # before calling this method.
@@ -1550,14 +1560,23 @@ def _add_auth_to_openapi_integration(
         editor.add_auth_to_method(api=api, path=self._path, method_name=self._method, auth=self.Auth)  # type: ignore[no-untyped-call]
 
 
-def _build_apigw_integration_uri(function, partition):  # type: ignore[no-untyped-def]
+def _build_apigw_integration_uri(function, partition, response_transfer_mode=None):  # type: ignore[no-untyped-def]
     function_arn = function.get_runtime_attr("arn")
+    # Use response-streaming-invocations path when ResponseTransferMode is RESPONSE_STREAM
+    # See: https://aws.amazon.com/blogs/compute/building-responsive-apis-with-amazon-api-gateway-response-streaming/
+    if response_transfer_mode == "RESPONSE_STREAM":
+        api_version = "2021-11-15"
+        invocation_path = "/response-streaming-invocations"
+    else:
+        api_version = "2015-03-31"
+        invocation_path = "/invocations"
+
     arn = (
         "arn:"
         + partition
-        + ":apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/"
+        + f":apigateway:${{AWS::Region}}:lambda:path/{api_version}/functions/"
         + make_shorthand(function_arn)
-        + "/invocations"
+        + invocation_path
     )
     # function_arn is always of the form {"Fn::GetAtt": ["<function_logical_id>", "Arn"]}.
     # We only want to check if the function logical id is a Py27UniStr instance.
diff --git a/samtranslator/open_api/open_api.py b/samtranslator/open_api/open_api.py
@@ -114,7 +114,14 @@ def is_integration_function_logical_id_match(self, path_name, method_name, logic
         return True
 
     def add_lambda_integration(  # type: ignore[no-untyped-def] # noqa: PLR0913
-        self, path, method, integration_uri, method_auth_config=None, api_auth_config=None, condition=None
+        self,
+        path,
+        method,
+        integration_uri,
+        method_auth_config=None,
+        api_auth_config=None,
+        condition=None,
+        invoke_mode=None,
     ):
         """
         Adds aws_proxy APIGW integration to the given path+method.
@@ -147,6 +154,9 @@ def add_lambda_integration(  # type: ignore[no-untyped-def] # noqa: PLR0913
             path_item[method][self._X_APIGW_INTEGRATION]["payloadFormatVersion"] = "2.0"
             path_item[method][self._X_APIGW_INTEGRATION]["uri"] = integration_uri
 
+            if invoke_mode:
+                path_item[method][self._X_APIGW_INTEGRATION]["invokeMode"] = invoke_mode
+
             if path == self._DEFAULT_PATH and method == self._X_ANY_METHOD:
                 path_item[method]["isDefaultRoute"] = True
 
diff --git a/samtranslator/schema/schema.json b/samtranslator/schema/schema.json
@@ -360664,6 +360664,15 @@
           "title": "RequestParameters",
           "type": "array"
         },
+        "ResponseTransferMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*CloudFormation compatibility*: This property is passed directly to the [`ResponseTransferMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-apigateway-method-integration.html#cfn-apigateway-method-integration-responsetransfermode) property of an `AWS::ApiGateway::Method Integration`\\.",
+          "title": "ResponseTransferMode"
+        },
         "RestApiId": {
           "anyOf": [
             {
diff --git a/samtranslator/swagger/swagger.py b/samtranslator/swagger/swagger.py
@@ -125,6 +125,7 @@ def add_lambda_integration(  # noqa: PLR0913
         method_auth_config: Dict[str, Any],
         api_auth_config: Dict[str, Any],
         condition: Optional[str] = None,
+        invoke_mode: Optional[Any] = None,
     ) -> None:
         """
         Adds aws_proxy APIGW integration to the given path+method.
@@ -150,6 +151,10 @@ def add_lambda_integration(  # noqa: PLR0913
             path_item[method][self._X_APIGW_INTEGRATION]["httpMethod"] = "POST"
             path_item[method][self._X_APIGW_INTEGRATION]["uri"] = _integration_uri
 
+            # When using RESPONSE_STREAM invoke mode, set responseTransferMode to STREAM
+            if invoke_mode == "RESPONSE_STREAM":
+                path_item[method][self._X_APIGW_INTEGRATION]["responseTransferMode"] = "STREAM"
+
             if (
                 method_auth_config.get("Authorizer") == "AWS_IAM"
                 or api_auth_config.get("DefaultAuthorizer") == "AWS_IAM"
diff --git a/schema_source/sam.schema.json b/schema_source/sam.schema.json
@@ -6103,6 +6103,15 @@
           "title": "RequestParameters",
           "type": "array"
         },
+        "ResponseTransferMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*CloudFormation compatibility*: This property is passed directly to the [`ResponseTransferMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-apigateway-method-integration.html#cfn-apigateway-method-integration-responsetransfermode) property of an `AWS::ApiGateway::Method Integration`\\.",
+          "title": "ResponseTransferMode"
+        },
         "RestApiId": {
           "anyOf": [
             {
diff --git a/tests/translator/input/api_with_invoke_mode.yaml b/tests/translator/input/api_with_invoke_mode.yaml
@@ -0,0 +1,20 @@
+Resources:
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/stream.zip
+      Handler: index.handler
+      Runtime: nodejs20.x
+      Events:
+        StreamApi:
+          Type: Api
+          Properties:
+            Path: /stream
+            Method: post
+            ResponseTransferMode: RESPONSE_STREAM
+        BufferedApi:
+          Type: Api
+          Properties:
+            Path: /buffered
+            Method: get
+            ResponseTransferMode: BUFFERED
diff --git a/tests/translator/output/api_with_invoke_mode.json b/tests/translator/output/api_with_invoke_mode.json
@@ -0,0 +1,162 @@
+{
+  "Resources": {
+    "MyFunction": {
+      "Properties": {
+        "Code": {
+          "S3Bucket": "sam-demo-bucket",
+          "S3Key": "stream.zip"
+        },
+        "Handler": "index.handler",
+        "Role": {
+          "Fn::GetAtt": [
+            "MyFunctionRole",
+            "Arn"
+          ]
+        },
+        "Runtime": "nodejs20.x",
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::Lambda::Function"
+    },
+    "MyFunctionBufferedApiPermissionProd": {
+      "Properties": {
+        "Action": "lambda:InvokeFunction",
+        "FunctionName": {
+          "Ref": "MyFunction"
+        },
+        "Principal": "apigateway.amazonaws.com",
+        "SourceArn": {
+          "Fn::Sub": [
+            "arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${__ApiId__}/${__Stage__}/GET/buffered",
+            {
+              "__ApiId__": {
+                "Ref": "ServerlessRestApi"
+              },
+              "__Stage__": "*"
+            }
+          ]
+        }
+      },
+      "Type": "AWS::Lambda::Permission"
+    },
+    "MyFunctionRole": {
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "sts:AssumeRole"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "Service": [
+                  "lambda.amazonaws.com"
+                ]
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "ManagedPolicyArns": [
+          "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+        ],
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::IAM::Role"
+    },
+    "MyFunctionStreamApiPermissionProd": {
+      "Properties": {
+        "Action": "lambda:InvokeFunction",
+        "FunctionName": {
+          "Ref": "MyFunction"
+        },
+        "Principal": "apigateway.amazonaws.com",
+        "SourceArn": {
+          "Fn::Sub": [
+            "arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${__ApiId__}/${__Stage__}/POST/stream",
+            {
+              "__ApiId__": {
+                "Ref": "ServerlessRestApi"
+              },
+              "__Stage__": "*"
+            }
+          ]
+        }
+      },
+      "Type": "AWS::Lambda::Permission"
+    },
+    "ServerlessRestApi": {
+      "Properties": {
+        "Body": {
+          "info": {
+            "title": {
+              "Ref": "AWS::StackName"
+            },
+            "version": "1.0"
+          },
+          "paths": {
+            "/buffered": {
+              "get": {
+                "responses": {},
+                "x-amazon-apigateway-integration": {
+                  "httpMethod": "POST",
+                  "type": "aws_proxy",
+                  "uri": {
+                    "Fn::Sub": "arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${MyFunction.Arn}/invocations"
+                  }
+                }
+              }
+            },
+            "/stream": {
+              "post": {
+                "responses": {},
+                "x-amazon-apigateway-integration": {
+                  "httpMethod": "POST",
+                  "responseTransferMode": "STREAM",
+                  "type": "aws_proxy",
+                  "uri": {
+                    "Fn::Sub": "arn:aws:apigateway:${AWS::Region}:lambda:path/2021-11-15/functions/${MyFunction.Arn}/response-streaming-invocations"
+                  }
+                }
+              }
+            }
+          },
+          "swagger": "2.0"
+        }
+      },
+      "Type": "AWS::ApiGateway::RestApi"
+    },
+    "ServerlessRestApiDeployment5a2e41cc40": {
+      "Properties": {
+        "Description": "RestApi deployment id: 5a2e41cc403a12dba16262d557905de64a61e2d1",
+        "RestApiId": {
+          "Ref": "ServerlessRestApi"
+        },
+        "StageName": "Stage"
+      },
+      "Type": "AWS::ApiGateway::Deployment"
+    },
+    "ServerlessRestApiProdStage": {
+      "Properties": {
+        "DeploymentId": {
+          "Ref": "ServerlessRestApiDeployment5a2e41cc40"
+        },
+        "RestApiId": {
+          "Ref": "ServerlessRestApi"
+        },
+        "StageName": "Prod"
+      },
+      "Type": "AWS::ApiGateway::Stage"
+    }
+  }
+}
diff --git a/tests/translator/output/aws-cn/api_with_invoke_mode.json b/tests/translator/output/aws-cn/api_with_invoke_mode.json
diff --git a/tests/translator/output/aws-us-gov/api_with_invoke_mode.json b/tests/translator/output/aws-us-gov/api_with_invoke_mode.json

Original file line number	Diff line number	Diff line change
`@@ -308,6 +308,7 @@ class ApiEventProperties(BaseModel):`
`308`	`308`	`"TimeoutInMillis",`
`309`	`309`	`["AWS::ApiGateway::Method.Integration", "TimeoutInMillis"],`
`310`	`310`	`)`
	`311`	`+ ResponseTransferMode: Optional[PassThroughProp] = apieventproperties("ResponseTransferMode")`
`311`	`312`
`312`	`313`
`313`	`314`	`class ApiEvent(BaseModel):`