Bases: BaseToolSpec
OpenAPI 工具。
该工具可用于解析 OpenAPI 规范中的端点和操作。使用 RequestsToolSpec 自动化对 openapi 服务器的请求。
Source code in llama-index-integrations/tools/llama-index-tools-openapi/llama_index/tools/openapi/base.py
10
11
12
13
14
15
16
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 | class OpenAPIToolSpec(BaseToolSpec):
"""
OpenAPI Tool.
This tool can be used to parse an OpenAPI spec for endpoints and operations
Use the RequestsToolSpec to automate requests to the openapi server
"""
spec_functions = ["load_openapi_spec"]
def __init__(self, spec: Optional[dict] = None, url: Optional[str] = None):
import yaml
if spec and url:
raise ValueError("Only provide one of OpenAPI dict or url")
elif spec:
pass
elif url:
response = requests.get(url).text
spec = yaml.safe_load(response)
else:
raise ValueError("You must provide a url or OpenAPI spec as a dict")
parsed_spec = self.process_api_spec(spec)
self.spec = Document(text=str(parsed_spec))
def load_openapi_spec(self) -> List[Document]:
"""
You are an AI agent specifically designed to retrieve information by making web requests to an API based on an OpenAPI specification.
Here's a step-by-step guide to assist you in answering questions:
1. Determine the base URL required for making the request
2. Identify the relevant paths necessary to address the question
3. Find the required parameters for making the request
4. Perform the necessary requests to obtain the answer
Returns:
Document: A List of Document objects.
"""
return [self.spec]
def process_api_spec(self, spec: dict) -> dict:
"""
Perform simplification and reduction on an OpenAPI specification.
The goal is to create a more concise and efficient representation
for retrieval purposes.
"""
def reduce_details(details: dict) -> dict:
reduced = {}
if details.get("description"):
reduced["description"] = details.get("description")
if details.get("parameters"):
reduced["parameters"] = [
param
for param in details.get("parameters", [])
if param.get("required")
]
if "200" in details["responses"]:
reduced["responses"] = details["responses"]["200"]
return reduced
def dereference_openapi(openapi_doc):
"""Dereferences a Swagger/OpenAPI document by resolving all $ref pointers."""
try:
import jsonschema
except ImportError:
raise ImportError(
"The jsonschema library is required to parse OpenAPI documents. "
"Please install it with `pip install jsonschema`."
)
resolver = jsonschema.RefResolver.from_schema(openapi_doc)
def _dereference(obj):
if isinstance(obj, dict):
if "$ref" in obj:
with resolver.resolving(obj["$ref"]) as resolved:
return _dereference(resolved)
return {k: _dereference(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_dereference(item) for item in obj]
else:
return obj
return _dereference(openapi_doc)
spec = dereference_openapi(spec)
endpoints = []
for route, operations in spec["paths"].items():
for operation, details in operations.items():
if operation in ["get", "post", "patch"]:
endpoint_name = f"{operation.upper()} {route}"
description = details.get("description")
endpoints.append(
(endpoint_name, description, reduce_details(details))
)
return {
"servers": spec["servers"],
"description": spec["info"].get("description"),
"endpoints": endpoints,
}
|
您是专门设计用于通过基于 OpenAPI 规范向 API 发起 Web 请求来检索信息的 AI 代理。
以下是协助您回答问题的分步指南：
-
确定发起请求所需的基础 URL
-
确定处理问题所需的相关路径
-
查找发起请求所需的参数
-
执行必要的请求以获取答案
返回值
Source code in llama-index-integrations/tools/llama-index-tools-openapi/llama_index/tools/openapi/base.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54 | def load_openapi_spec(self) -> List[Document]:
"""
You are an AI agent specifically designed to retrieve information by making web requests to an API based on an OpenAPI specification.
Here's a step-by-step guide to assist you in answering questions:
1. Determine the base URL required for making the request
2. Identify the relevant paths necessary to address the question
3. Find the required parameters for making the request
4. Perform the necessary requests to obtain the answer
Returns:
Document: A List of Document objects.
"""
return [self.spec]
|
process_api_spec(spec: dict) -> dict
对 OpenAPI 规范进行简化和缩减。
目标是创建一个更简洁高效的表示形式，以便于检索。
Source code in llama-index-integrations/tools/llama-index-tools-openapi/llama_index/tools/openapi/base.py
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 | def process_api_spec(self, spec: dict) -> dict:
"""
Perform simplification and reduction on an OpenAPI specification.
The goal is to create a more concise and efficient representation
for retrieval purposes.
"""
def reduce_details(details: dict) -> dict:
reduced = {}
if details.get("description"):
reduced["description"] = details.get("description")
if details.get("parameters"):
reduced["parameters"] = [
param
for param in details.get("parameters", [])
if param.get("required")
]
if "200" in details["responses"]:
reduced["responses"] = details["responses"]["200"]
return reduced
def dereference_openapi(openapi_doc):
"""Dereferences a Swagger/OpenAPI document by resolving all $ref pointers."""
try:
import jsonschema
except ImportError:
raise ImportError(
"The jsonschema library is required to parse OpenAPI documents. "
"Please install it with `pip install jsonschema`."
)
resolver = jsonschema.RefResolver.from_schema(openapi_doc)
def _dereference(obj):
if isinstance(obj, dict):
if "$ref" in obj:
with resolver.resolving(obj["$ref"]) as resolved:
return _dereference(resolved)
return {k: _dereference(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_dereference(item) for item in obj]
else:
return obj
return _dereference(openapi_doc)
spec = dereference_openapi(spec)
endpoints = []
for route, operations in spec["paths"].items():
for operation, details in operations.items():
if operation in ["get", "post", "patch"]:
endpoint_name = f"{operation.upper()} {route}"
description = details.get("description")
endpoints.append(
(endpoint_name, description, reduce_details(details))
)
return {
"servers": spec["servers"],
"description": spec["info"].get("description"),
"endpoints": endpoints,
}
|
