1  # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
  2  #
  3  # SPDX-License-Identifier: Apache-2.0
  4  
  5  from typing import Any
  6  
  7  from haystack import component, default_from_dict, default_to_dict
  8  from haystack.lazy_imports import LazyImport
  9  from haystack.utils import Secret
 10  
 11  with LazyImport("Run 'pip install openapi-llm'") as openapi_llm_imports:
 12      from openapi_llm.client.openapi import OpenAPIClient
 13  
 14  
 15  @component
 16  class OpenAPIConnector:
 17      """
 18      OpenAPIConnector enables direct invocation of REST endpoints defined in an OpenAPI specification.
 19  
 20      The OpenAPIConnector serves as a bridge between Haystack pipelines and any REST API that follows
 21      the OpenAPI(formerly Swagger) specification. It dynamically interprets the API specification and
 22      provides an interface for executing API operations. It is usually invoked by passing input
 23      arguments to it from a Haystack pipeline run method or by other components in a pipeline that
 24      pass input arguments to this component.
 25  
 26      Example:
 27      <!-- test-ignore -->
 28      ```python
 29      from haystack.utils import Secret
 30      from haystack.components.connectors.openapi import OpenAPIConnector
 31  
 32      serper_dev_token = Secret.from_env_var("SERPERDEV_API_KEY")
 33  
 34      def my_custom_config_factory():
 35          # Create and return a custom configuration for the OpenAPIClient
 36          pass
 37  
 38      connector = OpenAPIConnector(
 39          openapi_spec="https://bit.ly/serperdev_openapi",
 40          credentials=serper_dev_token,
 41          service_kwargs={"config_factory": my_custom_config_factory()}
 42      )
 43      response = connector.run(
 44          operation_id="search",
 45          arguments={"q": "Who was Nikola Tesla?"}
 46      )
 47      ```
 48      Note:
 49      - The `service_kwargs` argument is optional, it can be used to pass additional options to the OpenAPIClient.
 50  
 51      """
 52  
 53      def __init__(
 54          self, openapi_spec: str, credentials: Secret | None = None, service_kwargs: dict[str, Any] | None = None
 55      ) -> None:
 56          """
 57          Initialize the OpenAPIConnector with a specification and optional credentials.
 58  
 59          :param openapi_spec: URL, file path, or raw string of the OpenAPI specification
 60          :param credentials: Optional API key or credentials for the service wrapped in a Secret
 61          :param service_kwargs: Additional keyword arguments passed to OpenAPIClient.from_spec()
 62              For example, you can pass a custom config_factory or other configuration options.
 63          """
 64          openapi_llm_imports.check()
 65          self.openapi_spec = openapi_spec
 66          self.credentials = credentials
 67          self.service_kwargs = service_kwargs or {}
 68  
 69          self.client = OpenAPIClient.from_spec(
 70              openapi_spec=openapi_spec,
 71              credentials=credentials.resolve_value() if credentials else None,
 72              **self.service_kwargs,
 73          )
 74  
 75      def to_dict(self) -> dict[str, Any]:
 76          """
 77          Serialize this component to a dictionary.
 78          """
 79          return default_to_dict(
 80              self, openapi_spec=self.openapi_spec, credentials=self.credentials, service_kwargs=self.service_kwargs
 81          )
 82  
 83      @classmethod
 84      def from_dict(cls, data: dict[str, Any]) -> "OpenAPIConnector":
 85          """
 86          Deserialize this component from a dictionary.
 87          """
 88          return default_from_dict(cls, data)
 89  
 90      @component.output_types(response=dict[str, Any])
 91      def run(self, operation_id: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]:
 92          """
 93          Invokes a REST endpoint specified in the OpenAPI specification.
 94  
 95          :param operation_id: The operationId from the OpenAPI spec to invoke
 96          :param arguments: Optional parameters for the endpoint (query, path, or body parameters)
 97          :return: Dictionary containing the service response
 98          """
 99          payload = {"name": operation_id, "arguments": arguments or {}}
100  
101          # Invoke the endpoint using openapi-llm client
102          response = self.client.invoke(payload)
103          return {"response": response}