diff --git a/tests/aws/services/apigateway/test_apigateway_sqs.py b/tests/aws/services/apigateway/test_apigateway_sqs.py
index 30643d9c73c6e..4d05268621006 100644
--- a/tests/aws/services/apigateway/test_apigateway_sqs.py
+++ b/tests/aws/services/apigateway/test_apigateway_sqs.py
@@ -1,5 +1,6 @@
 import json
 import re
+import textwrap
 
 import pytest
 import requests
@@ -350,3 +351,144 @@ def get_sqs_message():
     message = retry(get_sqs_message, sleep=2, retries=10)
     snapshot.match("sqs-message-body", message["Body"])
     snapshot.match("sqs-message-attributes", message["MessageAttributes"])
+
+
+@markers.aws.validated
+@markers.snapshot.skip_snapshot_verify(
+    paths=[
+        # FIXME: those are minor parity gap in how we handle printing out VTL Map when they are nested inside bigger maps
+        "$..context.identity",
+        "$..context.requestOverride",
+        "$..context.responseOverride",
+        "$..requestOverride.header",
+        "$..requestOverride.path",
+        "$..requestOverride.querystring",
+        "$..responseOverride.header",
+        "$..responseOverride.path",
+        "$..responseOverride.status",
+    ]
+)
+def test_sqs_amz_json_protocol(
+    create_rest_apigw,
+    sqs_create_queue,
+    aws_client,
+    create_role_with_policy,
+    region_name,
+    account_id,
+    snapshot,
+    sqs_collect_messages,
+):
+    snapshot.add_transformer(snapshot.transform.sqs_api())
+    snapshot.add_transformers_list(
+        [
+            snapshot.transform.key_value("MD5OfBody"),
+            snapshot.transform.key_value("resourceId"),
+            snapshot.transform.key_value("extendedRequestId"),
+            snapshot.transform.key_value("requestTime"),
+            snapshot.transform.key_value("requestTimeEpoch", reference_replacement=False),
+            snapshot.transform.key_value("domainName"),
+            snapshot.transform.key_value("deploymentId"),
+            snapshot.transform.key_value("apiId"),
+            snapshot.transform.key_value("sourceIp"),
+        ]
+    )
+
+    # create target SQS stream
+    queue_name = f"queue-{short_uid()}"
+    queue_url = sqs_create_queue(QueueName=queue_name)
+
+    # create invocation role
+    _, role_arn = create_role_with_policy(
+        "Allow", "sqs:SendMessage", json.dumps(APIGATEWAY_ASSUME_ROLE_POLICY), "*"
+    )
+
+    api_id, _, root = create_rest_apigw(
+        name=f"test-api-{short_uid()}",
+        description="Test Integration with SQS",
+    )
+
+    resource_id = aws_client.apigateway.create_resource(
+        restApiId=api_id,
+        parentId=root,
+        pathPart="sqs",
+    )["id"]
+
+    aws_client.apigateway.put_method(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        authorizationType="NONE",
+    )
+
+    # we need to inline the JSON object because VTL does not handle newlines very well :/
+    context_template = textwrap.dedent(f"""
+        {{
+            "QueueUrl": "{queue_url}",
+            "MessageBody": "{{\\"context\\": {{#foreach( $key in $context.keySet() )\\"$key\\": \\"$context.get($key)\\"#if($foreach.hasNext),#end#end}},\\"identity\\": {{#foreach( $key in $context.identity.keySet() )\\"$key\\": \\"$context.identity.get($key)\\"#if($foreach.hasNext),#end#end}},\\"requestOverride\\": {{#foreach( $key in $context.requestOverride.keySet() )\\"$key\\": \\"$context.requestOverride.get($key)\\"#if($foreach.hasNext),#end#end}},\\"responseOverride\\": {{#foreach( $key in $context.responseOverride.keySet() )\\"$key\\": \\"$context.responseOverride.get($key)\\"#if($foreach.hasNext),#end#end}},\\"authorizer_keys\\": {{#foreach( $key in $context.authorizer.keySet() )\\"$key\\": \\"$util.escapeJavaScript($context.authorizer.get($key))\\"#if($foreach.hasNext),#end#end}}}}"}}
+    """)
+
+    aws_client.apigateway.put_integration(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        type="AWS",
+        integrationHttpMethod="POST",
+        uri=f"arn:aws:apigateway:{region_name}:sqs:path/{account_id}/{queue_name}",
+        credentials=role_arn,
+        requestParameters={
+            "integration.request.header.Content-Type": "'application/x-amz-json-1.0'",
+            "integration.request.header.X-Amz-Target": "'AmazonSQS.SendMessage'",
+        },
+        requestTemplates={"application/json": context_template},
+        passthroughBehavior="NEVER",
+    )
+
+    aws_client.apigateway.put_method_response(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        statusCode="200",
+        responseModels={"application/json": "Empty"},
+    )
+    aws_client.apigateway.put_method_response(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        statusCode="400",
+        responseModels={"application/json": "Empty"},
+    )
+
+    aws_client.apigateway.put_integration_response(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        statusCode="200",
+        responseTemplates={"application/json": '{"message": "great success!"}'},
+    )
+
+    aws_client.apigateway.put_integration_response(
+        restApiId=api_id,
+        resourceId=resource_id,
+        httpMethod="POST",
+        statusCode="400",
+        responseTemplates={"application/json": '{"message": "failure :("}'},
+        selectionPattern="400",
+    )
+
+    aws_client.apigateway.create_deployment(restApiId=api_id, stageName=TEST_STAGE_NAME)
+
+    invocation_url = api_invoke_url(api_id=api_id, stage=TEST_STAGE_NAME, path="/sqs")
+
+    def invoke_api(url):
+        _response = requests.post(url, headers={"User-Agent": "python/requests/tests"})
+        assert _response.ok
+        content = _response.json()
+        assert content == {"message": "great success!"}
+        return content
+
+    retry(invoke_api, sleep=2, retries=10, url=invocation_url)
+
+    messages = sqs_collect_messages(
+        queue_url=queue_url, expected=1, timeout=10, wait_time_seconds=5
+    )
+    snapshot.match("sqs-messages", messages)
diff --git a/tests/aws/services/apigateway/test_apigateway_sqs.snapshot.json b/tests/aws/services/apigateway/test_apigateway_sqs.snapshot.json
index c1be3da2b42e3..ef8e4017519ad 100644
--- a/tests/aws/services/apigateway/test_apigateway_sqs.snapshot.json
+++ b/tests/aws/services/apigateway/test_apigateway_sqs.snapshot.json
@@ -62,5 +62,62 @@
         }
       ]
     }
+  },
+  "tests/aws/services/apigateway/test_apigateway_sqs.py::test_sqs_amz_json_protocol": {
+    "recorded-date": "20-05-2025, 15:07:32",
+    "recorded-content": {
+      "sqs-messages": [
+        {
+          "MessageId": "<uuid:1>",
+          "ReceiptHandle": "<receipt-handle:1>",
+          "MD5OfBody": "<m-d5-of-body:1>",
+          "Body": {
+            "context": {
+              "resourceId": "<resource-id:1>",
+              "resourcePath": "/sqs",
+              "httpMethod": "POST",
+              "extendedRequestId": "<extended-request-id:1>",
+              "requestTime": "<request-time:1>",
+              "path": "/testing/sqs",
+              "accountId": "111111111111",
+              "protocol": "HTTP/1.1",
+              "requestOverride": "",
+              "stage": "testing",
+              "domainPrefix": "<api-id:1>",
+              "requestTimeEpoch": "request-time-epoch",
+              "requestId": "<uuid:2>",
+              "identity": "",
+              "domainName": "<domain-name:1>",
+              "deploymentId": "<deployment-id:1>",
+              "responseOverride": "",
+              "apiId": "<api-id:1>"
+            },
+            "identity": {
+              "cognitoIdentityPoolId": "",
+              "accountId": "",
+              "cognitoIdentityId": "",
+              "caller": "",
+              "sourceIp": "<source-ip:1>",
+              "principalOrgId": "",
+              "accessKey": "",
+              "cognitoAuthenticationType": "",
+              "cognitoAuthenticationProvider": "",
+              "userArn": "",
+              "userAgent": "python/requests/tests",
+              "user": ""
+            },
+            "requestOverride": {
+              "path": "",
+              "header": "",
+              "querystring": ""
+            },
+            "responseOverride": {
+              "header": ""
+            },
+            "authorizer_keys": {}
+          }
+        }
+      ]
+    }
   }
 }
diff --git a/tests/aws/services/apigateway/test_apigateway_sqs.validation.json b/tests/aws/services/apigateway/test_apigateway_sqs.validation.json
index a95f86484dac9..084035d99c099 100644
--- a/tests/aws/services/apigateway/test_apigateway_sqs.validation.json
+++ b/tests/aws/services/apigateway/test_apigateway_sqs.validation.json
@@ -1,4 +1,7 @@
 {
+  "tests/aws/services/apigateway/test_apigateway_sqs.py::test_sqs_amz_json_protocol": {
+    "last_validated_date": "2025-05-20T15:07:32+00:00"
+  },
   "tests/aws/services/apigateway/test_apigateway_sqs.py::test_sqs_aws_integration": {
     "last_validated_date": "2025-03-19T13:27:52+00:00"
   },