Skip to content

OpenAPILoader API Reference

langchain_openapi_tools.loader.OpenAPILoader

Loader for obtaining normalized OpenAPI specifications from various sources.

Supported sources include local JSON/YAML files, remote URLs, and in-memory dicts.

Source code in langchain_openapi_tools/loader.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class OpenAPILoader:
    """Loader for obtaining normalized OpenAPI specifications from various sources.

    Supported sources include local JSON/YAML files, remote URLs, and in-memory dicts.
    """

    def __init__(
        self,
        load_fn: Callable[[], dict[str, Any]],
        source_info: str,
    ) -> None:
        """Initialize loader with a fetch callable and a source description string.

        Args:
            load_fn: A zero-argument callable that returns a raw spec dictionary.
            source_info: Human-readable description of the source for logging.
        """
        self._load_fn = load_fn
        self._source_info = source_info

    @classmethod
    def from_url(
        cls,
        url: str,
        timeout: float = 30.0,
        headers: dict[str, str] | None = None,
    ) -> "OpenAPILoader":
        """Create an OpenAPILoader for a remote URL source.

        Args:
            url: HTTP or HTTPS URL of the OpenAPI specification.
            timeout: Request timeout in seconds (default: 30.0).
            headers: Optional dictionary of HTTP headers for the request.

        Returns:
            An OpenAPILoader instance configured for URL fetching.
        """

        def _fetch_from_url() -> dict[str, Any]:
            logger.info("Loading OpenAPI specification from URL: %s", url)
            req_headers = {"User-Agent": "langchain-openapi"}
            if headers:
                req_headers.update(headers)

            try:
                with httpx.Client(timeout=timeout, follow_redirects=True) as client:
                    response = client.get(url, headers=req_headers)
                    response.raise_for_status()
                    content = response.text
            except httpx.HTTPError as exc:
                logger.error("Failed to fetch OpenAPI spec from URL '%s': %s", url, exc)
                raise SpecLoadError(
                    f"Failed to fetch specification from URL '{url}': {exc}"
                ) from exc

            return parse_json_or_yaml(content)

        return cls(load_fn=_fetch_from_url, source_info=f"URL '{url}'")

    @classmethod
    def from_file(cls, file_path: str | Path) -> "OpenAPILoader":
        """Create an OpenAPILoader for a local JSON or YAML file source.

        Args:
            file_path: Path to the local OpenAPI specification file.

        Returns:
            An OpenAPILoader instance configured for file loading.
        """
        path = Path(file_path)

        def _read_from_file() -> dict[str, Any]:
            logger.info("Loading OpenAPI specification from file: %s", path)
            if not path.exists():
                logger.error("Specification file does not exist: %s", path)
                raise SpecLoadError(f"Specification file not found: '{path}'")

            if not path.is_file():
                logger.error("Path is not a regular file: %s", path)
                raise SpecLoadError(f"Path is not a regular file: '{path}'")

            try:
                content = path.read_text(encoding="utf-8")
            except OSError as exc:
                logger.error("Error reading specification file '%s': %s", path, exc)
                raise SpecLoadError(
                    f"Error reading specification file '{path}': {exc}"
                ) from exc

            return parse_json_or_yaml(content)

        return cls(load_fn=_read_from_file, source_info=f"file '{path}'")

    @classmethod
    def from_dict(cls, spec_dict: dict[str, Any]) -> "OpenAPILoader":
        """Create an OpenAPILoader from an in-memory dictionary.

        Args:
            spec_dict: Raw Python dictionary containing the OpenAPI specification.

        Returns:
            An OpenAPILoader instance configured for dictionary loading.
        """

        def _get_from_dict() -> dict[str, Any]:
            logger.info("Loading OpenAPI specification from in-memory dictionary")
            return spec_dict

        return cls(load_fn=_get_from_dict, source_info="in-memory dictionary")

    def load(self) -> OpenAPISpec:
        """Load, validate, and normalize the OpenAPI specification.

        Returns:
            An OpenAPISpec instance containing the normalized specification data.

        Raises:
            SpecLoadError: If loading from source or parsing JSON/YAML fails.
            InvalidSpecError: If required fields ('openapi', 'paths') are missing.
            UnsupportedVersionError: If the OpenAPI version is unsupported.
        """
        raw_data = self._load_fn()
        validate_raw_spec(raw_data)
        swagger_ver = str(raw_data.get("swagger", ""))
        if raw_data.get("swagger") == "2.0" or swagger_ver.startswith("2."):
            from langchain_openapi_tools.swagger import SwaggerNormalizer

            raw_data = SwaggerNormalizer(raw_data).normalize()
        spec = OpenAPISpec.from_dict(raw_data)

        logger.info(
            "Successfully loaded OpenAPI spec '%s' (OpenAPI version %s) from %s",
            spec.title,
            spec.version,
            self._source_info,
        )
        return spec

__init__(load_fn, source_info)

Initialize loader with a fetch callable and a source description string.

Parameters:

Name Type Description Default
load_fn Callable[[], dict[str, Any]]

A zero-argument callable that returns a raw spec dictionary.

 required
source_info str

Human-readable description of the source for logging.

 required
Source code in langchain_openapi_tools/loader.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    load_fn: Callable[[], dict[str, Any]],
    source_info: str,
) -> None:
    """Initialize loader with a fetch callable and a source description string.

    Args:
        load_fn: A zero-argument callable that returns a raw spec dictionary.
        source_info: Human-readable description of the source for logging.
    """
    self._load_fn = load_fn
    self._source_info = source_info

from_dict(spec_dict) classmethod

Create an OpenAPILoader from an in-memory dictionary.

Parameters:

Name Type Description Default
spec_dict dict[str, Any]

Raw Python dictionary containing the OpenAPI specification.

 required

Returns:

Type Description
OpenAPILoader

An OpenAPILoader instance configured for dictionary loading.
Source code in langchain_openapi_tools/loader.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@classmethod
def from_dict(cls, spec_dict: dict[str, Any]) -> "OpenAPILoader":
    """Create an OpenAPILoader from an in-memory dictionary.

    Args:
        spec_dict: Raw Python dictionary containing the OpenAPI specification.

    Returns:
        An OpenAPILoader instance configured for dictionary loading.
    """

    def _get_from_dict() -> dict[str, Any]:
        logger.info("Loading OpenAPI specification from in-memory dictionary")
        return spec_dict

    return cls(load_fn=_get_from_dict, source_info="in-memory dictionary")

from_file(file_path) classmethod

Create an OpenAPILoader for a local JSON or YAML file source.

Parameters:

Name Type Description Default
file_path str | Path

Path to the local OpenAPI specification file.

 required

Returns:

Type Description
OpenAPILoader

An OpenAPILoader instance configured for file loading.
Source code in langchain_openapi_tools/loader.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@classmethod
def from_file(cls, file_path: str | Path) -> "OpenAPILoader":
    """Create an OpenAPILoader for a local JSON or YAML file source.

    Args:
        file_path: Path to the local OpenAPI specification file.

    Returns:
        An OpenAPILoader instance configured for file loading.
    """
    path = Path(file_path)

    def _read_from_file() -> dict[str, Any]:
        logger.info("Loading OpenAPI specification from file: %s", path)
        if not path.exists():
            logger.error("Specification file does not exist: %s", path)
            raise SpecLoadError(f"Specification file not found: '{path}'")

        if not path.is_file():
            logger.error("Path is not a regular file: %s", path)
            raise SpecLoadError(f"Path is not a regular file: '{path}'")

        try:
            content = path.read_text(encoding="utf-8")
        except OSError as exc:
            logger.error("Error reading specification file '%s': %s", path, exc)
            raise SpecLoadError(
                f"Error reading specification file '{path}': {exc}"
            ) from exc

        return parse_json_or_yaml(content)

    return cls(load_fn=_read_from_file, source_info=f"file '{path}'")

from_url(url, timeout=30.0, headers=None) classmethod

Create an OpenAPILoader for a remote URL source.

Parameters:

Name Type Description Default
url str

HTTP or HTTPS URL of the OpenAPI specification.

 required
timeout float

Request timeout in seconds (default: 30.0).

 30.0
headers dict[str, str] | None

Optional dictionary of HTTP headers for the request.

 None

Returns:

Type Description
OpenAPILoader

An OpenAPILoader instance configured for URL fetching.
Source code in langchain_openapi_tools/loader.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@classmethod
def from_url(
    cls,
    url: str,
    timeout: float = 30.0,
    headers: dict[str, str] | None = None,
) -> "OpenAPILoader":
    """Create an OpenAPILoader for a remote URL source.

    Args:
        url: HTTP or HTTPS URL of the OpenAPI specification.
        timeout: Request timeout in seconds (default: 30.0).
        headers: Optional dictionary of HTTP headers for the request.

    Returns:
        An OpenAPILoader instance configured for URL fetching.
    """

    def _fetch_from_url() -> dict[str, Any]:
        logger.info("Loading OpenAPI specification from URL: %s", url)
        req_headers = {"User-Agent": "langchain-openapi"}
        if headers:
            req_headers.update(headers)

        try:
            with httpx.Client(timeout=timeout, follow_redirects=True) as client:
                response = client.get(url, headers=req_headers)
                response.raise_for_status()
                content = response.text
        except httpx.HTTPError as exc:
            logger.error("Failed to fetch OpenAPI spec from URL '%s': %s", url, exc)
            raise SpecLoadError(
                f"Failed to fetch specification from URL '{url}': {exc}"
            ) from exc

        return parse_json_or_yaml(content)

    return cls(load_fn=_fetch_from_url, source_info=f"URL '{url}'")

load()

Load, validate, and normalize the OpenAPI specification.

Returns:

Type Description
OpenAPISpec

An OpenAPISpec instance containing the normalized specification data.

Raises:

Type Description
SpecLoadError

If loading from source or parsing JSON/YAML fails.
InvalidSpecError

If required fields ('openapi', 'paths') are missing.
UnsupportedVersionError

If the OpenAPI version is unsupported.
Source code in langchain_openapi_tools/loader.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def load(self) -> OpenAPISpec:
    """Load, validate, and normalize the OpenAPI specification.

    Returns:
        An OpenAPISpec instance containing the normalized specification data.

    Raises:
        SpecLoadError: If loading from source or parsing JSON/YAML fails.
        InvalidSpecError: If required fields ('openapi', 'paths') are missing.
        UnsupportedVersionError: If the OpenAPI version is unsupported.
    """
    raw_data = self._load_fn()
    validate_raw_spec(raw_data)
    swagger_ver = str(raw_data.get("swagger", ""))
    if raw_data.get("swagger") == "2.0" or swagger_ver.startswith("2."):
        from langchain_openapi_tools.swagger import SwaggerNormalizer

        raw_data = SwaggerNormalizer(raw_data).normalize()
    spec = OpenAPISpec.from_dict(raw_data)

    logger.info(
        "Successfully loaded OpenAPI spec '%s' (OpenAPI version %s) from %s",
        spec.title,
        spec.version,
        self._source_info,
    )
    return spec