Source code for langchain_community.agent_toolkits.openapi.spec

"""Quick and dirty representation for OpenAPI specs."""

from dataclasses import dataclass
from typing import List, Tuple

from langchain_core.utils.json_schema import dereference_refs


[docs]@dataclass(frozen=True) class ReducedOpenAPISpec: """A reduced OpenAPI spec. This is a quick and dirty representation for OpenAPI specs. Parameters: servers: The servers in the spec. description: The description of the spec. endpoints: The endpoints in the spec. """ servers: List[dict] description: str endpoints: List[Tuple[str, str, dict]]
[docs]def reduce_openapi_spec(spec: dict, dereference: bool = True) -> ReducedOpenAPISpec: """Simplify/distill/minify a spec somehow. I want a smaller target for retrieval and (more importantly) I want smaller results from retrieval. I was hoping https://openapi.tools/ would have some useful bits to this end, but doesn't seem so. Args: spec: The OpenAPI spec. dereference: Whether to dereference the spec. Default is True. Returns: ReducedOpenAPISpec: The reduced OpenAPI spec. """ # 1. Consider only get, post, patch, put, delete endpoints. endpoints = [ (f"{operation_name.upper()} {route}", docs.get("description"), docs) for route, operation in spec["paths"].items() for operation_name, docs in operation.items() if operation_name in ["get", "post", "patch", "put", "delete"] ] # 2. Replace any refs so that complete docs are retrieved. # Note: probably want to do this post-retrieval, it blows up the size of the spec. if dereference: endpoints = [ (name, description, dereference_refs(docs, full_schema=spec)) for name, description, docs in endpoints ] # 3. Strip docs down to required request args + happy path response. def reduce_endpoint_docs(docs: dict) -> dict: out = {} if docs.get("description"): out["description"] = docs.get("description") if docs.get("parameters"): out["parameters"] = [ parameter for parameter in docs.get("parameters", []) if parameter.get("required") ] if "200" in docs["responses"]: out["responses"] = docs["responses"]["200"] if docs.get("requestBody"): out["requestBody"] = docs.get("requestBody") return out endpoints = [ (name, description, reduce_endpoint_docs(docs)) for name, description, docs in endpoints ] return ReducedOpenAPISpec( servers=spec["servers"], description=spec["info"].get("description", ""), endpoints=endpoints, )