diff --git a/CONTRIBUTING.rst b/CONTRIBUTING.rst
index fda8bfa0a2635..3ebb0fc9e9965 100644
--- a/CONTRIBUTING.rst
+++ b/CONTRIBUTING.rst
@@ -685,7 +685,6 @@ google                     amazon,apache.beam,apache.cassandra,cncf.kubernetes,f
 hashicorp                  google
 microsoft.azure            google,oracle
 mysql                      amazon,presto,trino,vertica
-opsgenie                   http
 postgres                   amazon
 salesforce                 tableau
 sftp                       ssh
diff --git a/airflow/providers/dependencies.json b/airflow/providers/dependencies.json
index 6779fec3f772f..ec082c1bb537b 100644
--- a/airflow/providers/dependencies.json
+++ b/airflow/providers/dependencies.json
@@ -67,9 +67,6 @@
     "trino",
     "vertica"
   ],
-  "opsgenie": [
-    "http"
-  ],
   "postgres": [
     "amazon"
   ],
diff --git a/airflow/providers/opsgenie/CHANGELOG.rst b/airflow/providers/opsgenie/CHANGELOG.rst
index 0ce5208bf66b0..4f2a6391168c7 100644
--- a/airflow/providers/opsgenie/CHANGELOG.rst
+++ b/airflow/providers/opsgenie/CHANGELOG.rst
@@ -19,6 +19,16 @@
 Changelog
 ---------
 
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+``OpsgenieAlertHook`` constructor does not take additional arguments or keyword arguments anymore.
+Changed the return type of ``OpsgenieAlertHook.get_conn`` to return an ``opsgenie_sdk.AlertApi`` object instead of a ``requests.Session`` object.
+Removed the ``OpsegnieAlertHook.execute`` method and replaced it with ``OpsegnieAlertHook.create_alert`` it now returns an
+``opsgenie_sdk.SuccessResponse`` object instead of an ``Any`` type.
+``OpsgenieAlertHook`` now takes ``visible_to`` instead of ``visibleTo`` key in the payload.
+``OpsgenieAlertHook`` now takes ``request_id`` instead of ``requestId`` key in the payload.
+
 2.0.1
 .....
 
diff --git a/airflow/providers/opsgenie/example_dags/__init__.py b/airflow/providers/opsgenie/example_dags/__init__.py
new file mode 100644
index 0000000000000..13a83393a9124
--- /dev/null
+++ b/airflow/providers/opsgenie/example_dags/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/airflow/providers/opsgenie/example_dags/example_opsgenie_alert.py b/airflow/providers/opsgenie/example_dags/example_opsgenie_alert.py
new file mode 100644
index 0000000000000..a39d356f9d78c
--- /dev/null
+++ b/airflow/providers/opsgenie/example_dags/example_opsgenie_alert.py
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from datetime import datetime
+
+from airflow import DAG
+from airflow.providers.opsgenie.operators.opsgenie_alert import OpsgenieAlertOperator
+
+with DAG(
+    dag_id="opsgenie_alert_operator_dag",
+    schedule_interval=None,
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+) as dag:
+
+    # [START howto_opsgenie_alert_operator]
+    opsgenie_alert_operator = OpsgenieAlertOperator(task_id="opsgenie_task", message="Hello World!")
+    # [END howto_opsgenie_alert_operator]
diff --git a/airflow/providers/opsgenie/hooks/opsgenie_alert.py b/airflow/providers/opsgenie/hooks/opsgenie_alert.py
index 3e1ef3d399b46..c3baae32a5411 100644
--- a/airflow/providers/opsgenie/hooks/opsgenie_alert.py
+++ b/airflow/providers/opsgenie/hooks/opsgenie_alert.py
@@ -17,16 +17,21 @@
 # under the License.
 #
 
-import json
-from typing import Any, Optional
+from typing import Optional
 
-import requests
+from opsgenie_sdk import (
+    AlertApi,
+    ApiClient,
+    Configuration,
+    CreateAlertPayload,
+    OpenApiException,
+    SuccessResponse,
+)
 
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.hooks.base import BaseHook
 
 
-class OpsgenieAlertHook(HttpHook):
+class OpsgenieAlertHook(BaseHook):
     """
     This hook allows you to post alerts to Opsgenie.
     Accepts a connection that has an Opsgenie API key as the connection's password.
@@ -46,46 +51,50 @@ class OpsgenieAlertHook(HttpHook):
     conn_type = 'opsgenie'
     hook_name = 'Opsgenie'
 
-    def __init__(self, opsgenie_conn_id: str = 'opsgenie_default', *args, **kwargs) -> None:
-        super().__init__(http_conn_id=opsgenie_conn_id, *args, **kwargs)  # type: ignore[misc]
+    def __init__(self, opsgenie_conn_id: str = 'opsgenie_default') -> None:
+        super().__init__()  # type: ignore[misc]
+        self.conn_id = opsgenie_conn_id
+        configuration = Configuration()
+        conn = self.get_connection(self.conn_id)
+        configuration.api_key['Authorization'] = conn.password
+        configuration.host = conn.host or 'https://api.opsgenie.com'
+        self.alert_api_instance = AlertApi(ApiClient(configuration))
 
     def _get_api_key(self) -> str:
-        """Get Opsgenie api_key for creating alert"""
-        conn = self.get_connection(self.http_conn_id)
-        api_key = conn.password
-        if not api_key:
-            raise AirflowException(
-                'Opsgenie API Key is required for this hook, please check your conn_id configuration.'
-            )
-        return api_key
-
-    def get_conn(self, headers: Optional[dict] = None) -> requests.Session:
         """
-        Overwrite HttpHook get_conn because this hook just needs base_url
-        and headers, and does not need generic params
+        Get the API key from the connection
 
-        :param headers: additional headers to be passed through as a dictionary
-        :type headers: dict
+        :return: API key
+        :rtype: str
         """
-        conn = self.get_connection(self.http_conn_id)
-        self.base_url = conn.host if conn.host else 'https://api.opsgenie.com'
-        session = requests.Session()
-        if headers:
-            session.headers.update(headers)
-        return session
-
-    def execute(self, payload: Optional[dict] = None) -> Any:
+        conn = self.get_connection(self.conn_id)
+        return conn.password
+
+    def get_conn(self) -> AlertApi:
+        """
+        Get the underlying AlertApi client
+
+        :return: AlertApi client
+        :rtype: opsgenie_sdk.AlertApi
         """
-        Execute the Opsgenie Alert call
+        return self.alert_api_instance
+
+    def create_alert(self, payload: Optional[dict] = None) -> SuccessResponse:
+        """
+        Create an alert on Opsgenie
 
         :param payload: Opsgenie API Create Alert payload values
             See https://docs.opsgenie.com/docs/alert-api#section-create-alert
         :type payload: dict
+        :return: api response
+        :rtype: opsgenie_sdk.SuccessResponse
         """
         payload = payload or {}
-        api_key = self._get_api_key()
-        return self.run(
-            endpoint='v2/alerts',
-            data=json.dumps(payload),
-            headers={'Content-Type': 'application/json', 'Authorization': f'GenieKey {api_key}'},
-        )
+
+        try:
+            create_alert_payload = CreateAlertPayload(**payload)
+            api_response = self.alert_api_instance.create_alert(create_alert_payload)
+            return api_response
+        except OpenApiException as e:
+            self.log.exception('Exception when sending alert to opsgenie with payload: %s', payload)
+            raise e
diff --git a/airflow/providers/opsgenie/operators/opsgenie_alert.py b/airflow/providers/opsgenie/operators/opsgenie_alert.py
index 47e5497efbf09..7554f88d2db24 100644
--- a/airflow/providers/opsgenie/operators/opsgenie_alert.py
+++ b/airflow/providers/opsgenie/operators/opsgenie_alert.py
@@ -32,6 +32,10 @@ class OpsgenieAlertOperator(BaseOperator):
     Each Opsgenie API key can be pre-configured to a team integration.
     You can override these defaults in this operator.
 
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:OpsgenieAlertOperator`
+
     :param opsgenie_conn_id: The name of the Opsgenie connection to use
     :type opsgenie_conn_id: str
     :param message: The Message of the Opsgenie alert (templated)
@@ -137,4 +141,4 @@ def _build_opsgenie_payload(self) -> Dict[str, Any]:
     def execute(self, context) -> None:
         """Call the OpsgenieAlertHook to post message"""
         self.hook = OpsgenieAlertHook(self.opsgenie_conn_id)
-        self.hook.execute(self._build_opsgenie_payload())
+        self.hook.create_alert(self._build_opsgenie_payload())
diff --git a/airflow/providers/opsgenie/provider.yaml b/airflow/providers/opsgenie/provider.yaml
index a392565cd58b7..eb2d5ba7de0f0 100644
--- a/airflow/providers/opsgenie/provider.yaml
+++ b/airflow/providers/opsgenie/provider.yaml
@@ -34,6 +34,8 @@ additional-dependencies:
 integrations:
   - integration-name: Opsgenie
     external-doc-url: https://www.opsgenie.com/
+    how-to-guide:
+      - /docs/apache-airflow-providers-opsgenie/operators/opsgenie_alert.rst
     logo: /integration-logos/opsgenie/Opsgenie.png
     tags: [service]
 
diff --git a/docs/apache-airflow-providers-opsgenie/index.rst b/docs/apache-airflow-providers-opsgenie/index.rst
index 6309537657bd6..ba647c1c51ac7 100644
--- a/docs/apache-airflow-providers-opsgenie/index.rst
+++ b/docs/apache-airflow-providers-opsgenie/index.rst
@@ -22,6 +22,12 @@
 Content
 -------
 
+.. toctree::
+    :maxdepth: 1
+    :caption: Guides
+
+    Operators <operators/index>
+
 .. toctree::
     :maxdepth: 1
     :caption: References
@@ -32,6 +38,7 @@ Content
     :maxdepth: 1
     :caption: Resources
 
+    Example DAGs <https://github.com/apache/airflow/tree/main/airflow/providers/opsgenie/example_dags>
     PyPI Repository <https://pypi.org/project/apache-airflow-providers-opsgenie/>
     Installing from sources <installing-providers-from-sources>
 
diff --git a/docs/apache-airflow-providers-opsgenie/operators/index.rst b/docs/apache-airflow-providers-opsgenie/operators/index.rst
new file mode 100644
index 0000000000000..ab140c6c44158
--- /dev/null
+++ b/docs/apache-airflow-providers-opsgenie/operators/index.rst
@@ -0,0 +1,26 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+
+Opsgenie Operators
+====================
+
+.. toctree::
+    :maxdepth: 1
+
+    opsgenie_alert
diff --git a/docs/apache-airflow-providers-opsgenie/operators/opsgenie_alert.rst b/docs/apache-airflow-providers-opsgenie/operators/opsgenie_alert.rst
new file mode 100644
index 0000000000000..7f6d094754050
--- /dev/null
+++ b/docs/apache-airflow-providers-opsgenie/operators/opsgenie_alert.rst
@@ -0,0 +1,33 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _howto/operator:OpsgenieAlertOperator:
+
+OpsgenieAlertOperator
+==========================
+
+Use the :class:`~airflow.providers.opsgenie.operators.opsgenie_alert.OpsgenieAlertOperator` to send an alert to opsgenie.
+
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+Send an alert to Opsgenie with a specific message.
+
+.. exampleinclude:: /../../airflow/providers/opsgenie/example_dags/example_opsgenie_alert.py
+    :language: python
+    :start-after: [START howto_opsgenie_alert_operator]
+    :end-before: [END howto_opsgenie_alert_operator]
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index b196744b5b241..6d065788ccb99 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -5,6 +5,7 @@ AgentKey
 Airbnb
 Airbyte
 AirflowException
+AlertApi
 Alibaba
 Alphasort
 Analytics
@@ -356,6 +357,7 @@ Subpackages
 Subpath
 SubscriberClient
 Subtasks
+SuccessResponse
 Systemd
 TCP
 TLS
diff --git a/setup.py b/setup.py
index 744f8679d24e6..cea42f5031a45 100644
--- a/setup.py
+++ b/setup.py
@@ -405,6 +405,9 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
 odbc = [
     'pyodbc',
 ]
+opsgenie = [
+    'opsgenie-sdk>=2.1.5',
+]
 oracle = [
     'cx_Oracle>=5.1.2',
 ]
@@ -641,7 +644,7 @@ def write_version(filename: str = os.path.join(*[my_dir, "airflow", "git_version
     'neo4j': neo4j,
     'odbc': odbc,
     'openfaas': [],
-    'opsgenie': http_provider,
+    'opsgenie': opsgenie,
     'oracle': oracle,
     'pagerduty': pagerduty,
     'papermill': papermill,
diff --git a/tests/providers/opsgenie/hooks/test_opsgenie_alert.py b/tests/providers/opsgenie/hooks/test_opsgenie_alert.py
index 89f8b40ea3ed5..a03b350545d14 100644
--- a/tests/providers/opsgenie/hooks/test_opsgenie_alert.py
+++ b/tests/providers/opsgenie/hooks/test_opsgenie_alert.py
@@ -16,13 +16,13 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-import json
 import unittest
+from unittest import mock
 
 import pytest
-import requests_mock
+from opsgenie_sdk import AlertApi, CreateAlertPayload
+from opsgenie_sdk.exceptions import AuthenticationException
 
-from airflow.exceptions import AirflowException
 from airflow.models import Connection
 from airflow.providers.opsgenie.hooks.opsgenie_alert import OpsgenieAlertHook
 from airflow.utils import db
@@ -45,7 +45,7 @@ class TestOpsgenieAlertHook(unittest.TestCase):
             {'id': '80564037-1984-4f38-b98e-8a1f662df552', 'type': 'schedule'},
             {'name': 'First Responders Schedule', 'type': 'schedule'},
         ],
-        'visibleTo': [
+        'visible_to': [
             {'id': '4513b7ea-3b91-438f-b7e4-e3e54af9147c', 'type': 'team'},
             {'name': 'rocket_team', 'type': 'team'},
             {'id': 'bb4d9938-c3c2-455d-aaab-727aa701c0d8', 'type': 'user'},
@@ -63,7 +63,7 @@ class TestOpsgenieAlertHook(unittest.TestCase):
     _mock_success_response_body = {
         "result": "Request will be processed",
         "took": 0.302,
-        "requestId": "43a29c5c-3dbf-4fa4-9c26-f4f71023e120",
+        "request_id": "43a29c5c-3dbf-4fa4-9c26-f4f71023e120",
     }
 
     def setUp(self):
@@ -83,34 +83,36 @@ def test_get_api_key(self):
 
     def test_get_conn_defaults_host(self):
         hook = OpsgenieAlertHook()
-        hook.get_conn()
-        assert 'https://api.opsgenie.com' == hook.base_url
+        assert 'https://api.opsgenie.com' == hook.get_conn().api_client.configuration.host
 
-    @requests_mock.mock()
-    def test_call_with_success(self, m):
-        hook = OpsgenieAlertHook(opsgenie_conn_id=self.conn_id)
-        m.post(self.opsgenie_alert_endpoint, status_code=202, json=self._mock_success_response_body)
-        resp = hook.execute(payload=self._payload)
-        assert resp.status_code == 202
-        assert resp.json() == self._mock_success_response_body
+    def test_get_conn_custom_host(self):
+        conn_id = 'custom_host_opsgenie_test'
+        db.merge_conn(
+            Connection(
+                conn_id=conn_id,
+                conn_type='opsgenie',
+                host='https://app.eu.opsgenie.com',
+                password='eb243592-faa2-4ba2-a551q-1afdf565c889',
+            )
+        )
 
-    @requests_mock.mock()
-    def test_api_key_set(self, m):
+        hook = OpsgenieAlertHook(conn_id)
+        assert 'https://app.eu.opsgenie.com' == hook.get_conn().api_client.configuration.host
+
+    def test_verify_api_key_set(self):
         hook = OpsgenieAlertHook(opsgenie_conn_id=self.conn_id)
-        m.post(self.opsgenie_alert_endpoint, status_code=202, json=self._mock_success_response_body)
-        resp = hook.execute(payload=self._payload)
-        assert resp.request.headers.get('Authorization') == 'GenieKey eb243592-faa2-4ba2-a551q-1afdf565c889'
+        assert (
+            hook.alert_api_instance.api_client.configuration.api_key.get('Authorization', None)
+            == 'eb243592-faa2-4ba2-a551q-1afdf565c889'
+        )
 
-    @requests_mock.mock()
-    def test_api_key_not_set(self, m):
+    def test_api_key_not_set(self):
         hook = OpsgenieAlertHook()
-        m.post(self.opsgenie_alert_endpoint, status_code=202, json=self._mock_success_response_body)
-        with pytest.raises(AirflowException):
-            hook.execute(payload=self._payload)
+        with pytest.raises(AuthenticationException):
+            hook.create_alert(payload=self._payload)
 
-    @requests_mock.mock()
-    def test_payload_set(self, m):
+    @mock.patch.object(AlertApi, 'create_alert')
+    def test_payload(self, create_alert_mock):
         hook = OpsgenieAlertHook(opsgenie_conn_id=self.conn_id)
-        m.post(self.opsgenie_alert_endpoint, status_code=202, json=self._mock_success_response_body)
-        resp = hook.execute(payload=self._payload)
-        assert json.loads(resp.request.body) == self._payload
+        hook.create_alert(payload=self._payload)
+        create_alert_mock.assert_called_once_with(CreateAlertPayload(**self._payload))
