diff --git a/docs/pydoc/config/connectors.yml b/docs/pydoc/config/connectors.yml
index b53b4bb80f..cd9ce98a02 100644
--- a/docs/pydoc/config/connectors.yml
+++ b/docs/pydoc/config/connectors.yml
@@ -1,7 +1,7 @@
 loaders:
   - type: haystack_pydoc_tools.loaders.CustomPythonLoader
     search_path: [../../../haystack/components/connectors]
-    modules: ["openapi_service"]
+    modules: ["openapi_service", "openapi"]
     ignore_when_discovered: ["__init__"]
 processors:
   - type: filter
diff --git a/haystack/components/connectors/__init__.py b/haystack/components/connectors/__init__.py
index b9fbe418aa..c5bbf4537a 100644
--- a/haystack/components/connectors/__init__.py
+++ b/haystack/components/connectors/__init__.py
@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from haystack.components.connectors.openapi import OpenAPIConnector
 from haystack.components.connectors.openapi_service import OpenAPIServiceConnector
 
-__all__ = ["OpenAPIServiceConnector"]
+__all__ = ["OpenAPIServiceConnector", "OpenAPIConnector"]
diff --git a/haystack/components/connectors/openapi.py b/haystack/components/connectors/openapi.py
new file mode 100644
index 0000000000..55e6024534
--- /dev/null
+++ b/haystack/components/connectors/openapi.py
@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Any, Dict, Optional
+
+from haystack import component, default_from_dict, default_to_dict, logging
+from haystack.lazy_imports import LazyImport
+from haystack.utils import Secret, deserialize_secrets_inplace
+
+with LazyImport("Run 'pip install openapi-llm'") as openapi_llm_imports:
+    from openapi_llm.client.openapi import OpenAPIClient
+
+logger = logging.getLogger(__name__)
+
+
+@component
+class OpenAPIConnector:
+    """
+    OpenAPIConnector enables direct invocation of REST endpoints defined in an OpenAPI specification.
+
+    The OpenAPIConnector serves as a bridge between Haystack pipelines and any REST API that follows
+    the OpenAPI(formerly Swagger) specification. It dynamically interprets the API specification and
+    provides an interface for executing API operations. It is usually invoked by passing input
+    arguments to it from a Haystack pipeline run method or by other components in a pipeline that
+    pass input arguments to this component.
+
+    Example:
+    ```python
+    from haystack.utils import Secret
+    from haystack.components.connectors.openapi import OpenAPIConnector
+
+    connector = OpenAPIConnector(
+        openapi_spec="https://bit.ly/serperdev_openapi",
+        credentials=Secret.from_env_var("SERPERDEV_API_KEY"),
+        service_kwargs={"config_factory": my_custom_config_factory}
+    )
+    response = connector.run(
+        operation_id="search",
+        parameters={"q": "Who was Nikola Tesla?"}
+    )
+    ```
+    Note:
+    - The `parameters` argument is required for this component.
+    - The `service_kwargs` argument is optional, it can be used to pass additional options to the OpenAPIClient.
+
+    """
+
+    def __init__(
+        self, openapi_spec: str, credentials: Optional[Secret] = None, service_kwargs: Optional[Dict[str, Any]] = None
+    ):
+        """
+        Initialize the OpenAPIConnector with a specification and optional credentials.
+
+        :param openapi_spec: URL, file path, or raw string of the OpenAPI specification
+        :param credentials: Optional API key or credentials for the service wrapped in a Secret
+        :param service_kwargs: Additional keyword arguments passed to OpenAPIClient.from_spec()
+            For example, you can pass a custom config_factory or other configuration options.
+        """
+        openapi_llm_imports.check()
+        self.openapi_spec = openapi_spec
+        self.credentials = credentials
+        self.service_kwargs = service_kwargs or {}
+
+        self.client = OpenAPIClient.from_spec(
+            openapi_spec=openapi_spec,
+            credentials=credentials.resolve_value() if credentials else None,
+            **self.service_kwargs,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Serialize this component to a dictionary.
+        """
+        return default_to_dict(
+            self,
+            openapi_spec=self.openapi_spec,
+            credentials=self.credentials.to_dict() if self.credentials else None,
+            service_kwargs=self.service_kwargs,
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "OpenAPIConnector":
+        """
+        Deserialize this component from a dictionary.
+        """
+        deserialize_secrets_inplace(data["init_parameters"], keys=["credentials"])
+        return default_from_dict(cls, data)
+
+    @component.output_types(response=Dict[str, Any])
+    def run(self, operation_id: str, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """
+        Invokes a REST endpoint specified in the OpenAPI specification.
+
+        :param operation_id: The operationId from the OpenAPI spec to invoke
+        :param parameters: Optional parameters for the endpoint (query, path, or body parameters)
+        :return: Dictionary containing the service response
+        """
+        payload = {"name": operation_id, "arguments": arguments or {}}
+
+        # Invoke the endpoint using openapi-llm client
+        response = self.client.invoke(payload)
+        return {"response": response}
diff --git a/pyproject.toml b/pyproject.toml
index 32ae00497c..92ba126ba9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -58,6 +58,7 @@ dependencies = [
   "numpy",
   "python-dateutil",
   "jsonschema",             # JsonSchemaValidator, Tool
+  "openapi-llm>=0.4.1",            # OpenAPIConnector
   "haystack-experimental",
 ]
 
diff --git a/releasenotes/notes/add-openapi-connector-ebaa97cfa95b6c3e.yaml b/releasenotes/notes/add-openapi-connector-ebaa97cfa95b6c3e.yaml
new file mode 100644
index 0000000000..bfb44ce9e7
--- /dev/null
+++ b/releasenotes/notes/add-openapi-connector-ebaa97cfa95b6c3e.yaml
@@ -0,0 +1,21 @@
+---
+features:
+  - |
+    Introduced the OpenAPIConnector component, enabling direct invocation of REST endpoints as specified in an OpenAPI specification.
+    This component is designed for direct REST endpoint invocation without LLM-generated payloads, users needs
+    to pass the run parameters explicitly.
+
+    Example:
+    ```python
+    from haystack.utils import Secret
+    from haystack.components.connectors.openapi import OpenAPIConnector
+
+    connector = OpenAPIConnector(
+        openapi_spec="https://bit.ly/serperdev_openapi",
+        credentials=Secret.from_env_var("SERPERDEV_API_KEY"),
+    )
+    response = connector.run(
+        operation_id="search",
+        parameters={"q": "Who was Nikola Tesla?"}
+    )
+    ```
diff --git a/test/components/connectors/test_openapi_connector.py b/test/components/connectors/test_openapi_connector.py
new file mode 100644
index 0000000000..48cdfade64
--- /dev/null
+++ b/test/components/connectors/test_openapi_connector.py
@@ -0,0 +1,205 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+from unittest.mock import Mock, patch
+
+import pytest
+from haystack import Pipeline
+from haystack.utils import Secret
+from haystack.components.connectors.openapi import OpenAPIConnector
+
+# Mock OpenAPI spec for testing
+MOCK_OPENAPI_SPEC = """
+openapi: 3.0.0
+info:
+  title: Test API
+  version: 1.0.0
+paths:
+  /search:
+    get:
+      operationId: search
+      parameters:
+        - name: q
+          in: query
+          required: true
+          schema:
+            type: string
+"""
+
+
+@pytest.fixture
+def mock_client():
+    with patch("haystack.components.connectors.openapi.OpenAPIClient") as mock:
+        client_instance = Mock()
+        mock.from_spec.return_value = client_instance
+        yield client_instance
+
+
+class TestOpenAPIConnector:
+    def test_init(self, mock_client):
+        # Test initialization with credentials and service_kwargs
+        service_kwargs = {"allowed_operations": ["search"]}
+        connector = OpenAPIConnector(
+            openapi_spec=MOCK_OPENAPI_SPEC, credentials=Secret.from_token("test-token"), service_kwargs=service_kwargs
+        )
+        assert connector.openapi_spec == MOCK_OPENAPI_SPEC
+        assert connector.credentials.resolve_value() == "test-token"
+        assert connector.service_kwargs == service_kwargs
+
+        # Test initialization without credentials and service_kwargs
+        connector = OpenAPIConnector(openapi_spec=MOCK_OPENAPI_SPEC)
+        assert connector.credentials is None
+        assert connector.service_kwargs == {}
+
+    def test_to_dict(self, monkeypatch):
+        monkeypatch.setenv("ENV_VAR", "test-api-key")
+        service_kwargs = {"allowed_operations": ["search"]}
+        connector = OpenAPIConnector(
+            openapi_spec=MOCK_OPENAPI_SPEC, credentials=Secret.from_env_var("ENV_VAR"), service_kwargs=service_kwargs
+        )
+        serialized = connector.to_dict()
+        assert serialized == {
+            "type": "haystack.components.connectors.openapi.OpenAPIConnector",
+            "init_parameters": {
+                "openapi_spec": MOCK_OPENAPI_SPEC,
+                "credentials": {"env_vars": ["ENV_VAR"], "type": "env_var", "strict": True},
+                "service_kwargs": service_kwargs,
+            },
+        }
+
+    def test_from_dict(self, monkeypatch):
+        monkeypatch.setenv("ENV_VAR", "test-api-key")
+        service_kwargs = {"allowed_operations": ["search"]}
+        data = {
+            "type": "haystack.components.connectors.openapi.OpenAPIConnector",
+            "init_parameters": {
+                "openapi_spec": MOCK_OPENAPI_SPEC,
+                "credentials": {"env_vars": ["ENV_VAR"], "type": "env_var", "strict": True},
+                "service_kwargs": service_kwargs,
+            },
+        }
+        connector = OpenAPIConnector.from_dict(data)
+        assert connector.openapi_spec == MOCK_OPENAPI_SPEC
+        assert connector.credentials == Secret.from_env_var("ENV_VAR")
+        assert connector.service_kwargs == service_kwargs
+
+    def test_run(self, mock_client):
+        service_kwargs = {"allowed_operations": ["search"]}
+        connector = OpenAPIConnector(
+            openapi_spec=MOCK_OPENAPI_SPEC, credentials=Secret.from_token("test-token"), service_kwargs=service_kwargs
+        )
+
+        # Mock the response from the client
+        mock_client.invoke.return_value = {"results": ["test result"]}
+
+        # Test with arguments
+        response = connector.run(operation_id="search", arguments={"q": "test query"})
+        mock_client.invoke.assert_called_with({"name": "search", "arguments": {"q": "test query"}})
+        assert response == {"response": {"results": ["test result"]}}
+
+        # Test without arguments
+        response = connector.run(operation_id="search")
+        mock_client.invoke.assert_called_with({"name": "search", "arguments": {}})
+
+    def test_in_pipeline(self, mock_client):
+        mock_client.invoke.return_value = {"results": ["test result"]}
+
+        connector = OpenAPIConnector(openapi_spec=MOCK_OPENAPI_SPEC, credentials=Secret.from_token("test-token"))
+
+        pipe = Pipeline()
+        pipe.add_component("api", connector)
+
+        # Test pipeline execution
+        results = pipe.run(data={"api": {"operation_id": "search", "arguments": {"q": "test query"}}})
+
+        assert results == {"api": {"response": {"results": ["test result"]}}}
+
+    def test_from_dict_fail_wo_env_var(self, monkeypatch):
+        monkeypatch.delenv("ENV_VAR", raising=False)
+        data = {
+            "type": "haystack.components.connectors.openapi.OpenAPIConnector",
+            "init_parameters": {
+                "openapi_spec": MOCK_OPENAPI_SPEC,
+                "credentials": {"env_vars": ["ENV_VAR"], "type": "env_var", "strict": True},
+            },
+        }
+        with pytest.raises(ValueError, match="None of the .* environment variables are set"):
+            OpenAPIConnector.from_dict(data)
+
+    def test_serde_in_pipeline(self, monkeypatch):
+        """
+        Test serialization/deserialization of OpenAPIConnector in a Pipeline,
+        including detailed dictionary validation
+        """
+        monkeypatch.setenv("API_KEY", "test-api-key")
+
+        # Create connector with specific configuration
+        connector = OpenAPIConnector(
+            openapi_spec=MOCK_OPENAPI_SPEC,
+            credentials=Secret.from_env_var("API_KEY"),
+            service_kwargs={"allowed_operations": ["search"]},
+        )
+
+        # Create and configure pipeline
+        pipeline = Pipeline()
+        pipeline.add_component("api", connector)
+
+        # Get pipeline dictionary and verify its structure
+        pipeline_dict = pipeline.to_dict()
+        assert pipeline_dict == {
+            "metadata": {},
+            "max_runs_per_component": 100,
+            "components": {
+                "api": {
+                    "type": "haystack.components.connectors.openapi.OpenAPIConnector",
+                    "init_parameters": {
+                        "openapi_spec": MOCK_OPENAPI_SPEC,
+                        "credentials": {"env_vars": ["API_KEY"], "type": "env_var", "strict": True},
+                        "service_kwargs": {"allowed_operations": ["search"]},
+                    },
+                }
+            },
+            "connections": [],
+        }
+
+        # Test YAML serialization/deserialization
+        pipeline_yaml = pipeline.dumps()
+        new_pipeline = Pipeline.loads(pipeline_yaml)
+        assert new_pipeline == pipeline
+
+        # Verify the loaded pipeline's connector has the same configuration
+        loaded_connector = new_pipeline.get_component("api")
+        assert loaded_connector.openapi_spec == connector.openapi_spec
+        assert loaded_connector.credentials == connector.credentials
+        assert loaded_connector.service_kwargs == connector.service_kwargs
+
+
+@pytest.mark.integration
+class TestOpenAPIConnectorIntegration:
+    @pytest.mark.skipif(
+        not os.environ.get("SERPERDEV_API_KEY", None),
+        reason="Export an env var called SERPERDEV_API_KEY to run this test.",
+    )
+    @pytest.mark.integration
+    def test_serper_dev_integration(self):
+        component = OpenAPIConnector(
+            openapi_spec="https://bit.ly/serperdev_openapi", credentials=Secret.from_env_var("SERPERDEV_API_KEY")
+        )
+        response = component.run(operation_id="search", arguments={"q": "Who was Nikola Tesla?"})
+        assert isinstance(response, dict)
+        assert "response" in response
+
+    @pytest.mark.skipif(
+        not os.environ.get("GITHUB_TOKEN", None), reason="Export an env var called GITHUB_TOKEN to run this test."
+    )
+    @pytest.mark.integration
+    def test_github_api_integration(self):
+        component = OpenAPIConnector(
+            openapi_spec="https://raw.githubusercontent.com/github/rest-api-description/main/descriptions/api.github.com/api.github.com.json",
+            credentials=Secret.from_env_var("GITHUB_TOKEN"),
+        )
+        response = component.run(operation_id="search_repos", arguments={"q": "deepset-ai"})
+        assert isinstance(response, dict)
+        assert "response" in response