Spaces:
Runtime error
Runtime error
File size: 1,630 Bytes
35b22df |
1 2 3 4 5 6 7 8 9 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 |
"""Make.com API wrapper.
Currently cannot load documents.
"""
from typing import Any, List, Optional
import requests
from gpt_index.readers.base import BaseReader
from gpt_index.readers.schema.base import Document
from gpt_index.response.schema import Response, SourceNode
class MakeWrapper(BaseReader):
"""Make reader."""
def load_data(self, *args: Any, **load_kwargs: Any) -> List[Document]:
"""Load data from the input directory.
NOTE: This is not implemented.
"""
raise NotImplementedError("Cannot load documents from Make.com API.")
def pass_response_to_webhook(
self, webhook_url: str, response: Response, query: Optional[str] = None
) -> None:
"""Pass response object to webhook.
Args:
webhook_url (str): Webhook URL.
response (Response): Response object.
query (Optional[str]): Query. Defaults to None.
"""
response_text = response.response
source_nodes = [n.to_dict() for n in response.source_nodes]
json_dict = {
"response": response_text,
"source_nodes": source_nodes,
"query": query,
}
r = requests.post(webhook_url, json=json_dict)
r.raise_for_status()
if __name__ == "__main__":
wrapper = MakeWrapper()
test_response = Response(
response="test response",
source_nodes=[SourceNode(source_text="test source", doc_id="test id")],
)
wrapper.pass_response_to_webhook(
"https://hook.us1.make.com/asdfadsfasdfasdfd",
test_response,
"Test query",
)
|