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
154
155
156
157
158
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,
        source_url: str | None = None,
    ) -> 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.
            source_url: Optional absolute URL the specification was fetched
                from. Preserved on the resulting ``OpenAPISpec`` so relative
                ``servers`` entries and specs with no ``servers`` block can be
                resolved to absolute request URLs.
        """
        self._load_fn = load_fn
        self._source_info = source_info
        self._source_url = source_url

    @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}'",
            source_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)
        spec = OpenAPISpec.from_dict(raw_data, source_url=self._source_url)

        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, source_url=None)

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_url str | None

Optional absolute URL the specification was fetched from. Preserved on the resulting OpenAPISpec so relative servers entries and specs with no servers block can be resolved to absolute request URLs.

 None
Source code in langchain_openapi_tools/loader.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(
    self,
    load_fn: Callable[[], dict[str, Any]],
    source_info: str,
    source_url: str | None = None,
) -> 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.
        source_url: Optional absolute URL the specification was fetched
            from. Preserved on the resulting ``OpenAPISpec`` so relative
            ``servers`` entries and specs with no ``servers`` block can be
            resolved to absolute request URLs.
    """
    self._load_fn = load_fn
    self._source_info = source_info
    self._source_url = source_url

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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@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 
 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
@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 
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
@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}'",
        source_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 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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)
    spec = OpenAPISpec.from_dict(raw_data, source_url=self._source_url)

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