Spaces:
Runtime error
Runtime error
File size: 4,961 Bytes
58d33f0 |
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 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 |
# flake8: noqa
"""Tools for making requests to an API endpoint."""
import json
from typing import Any, Dict
from pydantic import BaseModel
from langchain.requests import RequestsWrapper
from langchain.tools.base import BaseTool
def _parse_input(text: str) -> Dict[str, Any]:
"""Parse the json string into a dict."""
return json.loads(text)
class BaseRequestsTool(BaseModel):
"""Base class for requests tools."""
requests_wrapper: RequestsWrapper
class RequestsGetTool(BaseRequestsTool, BaseTool):
"""Tool for making a GET request to an API endpoint."""
name = "requests_get"
description = "A portal to the internet. Use this when you need to get specific content from a website. Input should be a url (i.e. https://www.google.com). The output will be the text response of the GET request."
def _run(self, url: str) -> str:
"""Run the tool."""
return self.requests_wrapper.get(url)
async def _arun(self, url: str) -> str:
"""Run the tool asynchronously."""
return await self.requests_wrapper.aget(url)
class RequestsPostTool(BaseRequestsTool, BaseTool):
"""Tool for making a POST request to an API endpoint."""
name = "requests_post"
description = """Use this when you want to POST to a website.
Input should be a json string with two keys: "url" and "data".
The value of "url" should be a string, and the value of "data" should be a dictionary of
key-value pairs you want to POST to the url.
Be careful to always use double quotes for strings in the json string
The output will be the text response of the POST request.
"""
def _run(self, text: str) -> str:
"""Run the tool."""
try:
data = _parse_input(text)
return self.requests_wrapper.post(data["url"], data["data"])
except Exception as e:
return repr(e)
async def _arun(self, text: str) -> str:
"""Run the tool asynchronously."""
try:
data = _parse_input(text)
return await self.requests_wrapper.apost(data["url"], data["data"])
except Exception as e:
return repr(e)
class RequestsPatchTool(BaseRequestsTool, BaseTool):
"""Tool for making a PATCH request to an API endpoint."""
name = "requests_patch"
description = """Use this when you want to PATCH to a website.
Input should be a json string with two keys: "url" and "data".
The value of "url" should be a string, and the value of "data" should be a dictionary of
key-value pairs you want to PATCH to the url.
Be careful to always use double quotes for strings in the json string
The output will be the text response of the PATCH request.
"""
def _run(self, text: str) -> str:
"""Run the tool."""
try:
data = _parse_input(text)
return self.requests_wrapper.patch(data["url"], data["data"])
except Exception as e:
return repr(e)
async def _arun(self, text: str) -> str:
"""Run the tool asynchronously."""
try:
data = _parse_input(text)
return await self.requests_wrapper.apatch(data["url"], data["data"])
except Exception as e:
return repr(e)
class RequestsPutTool(BaseRequestsTool, BaseTool):
"""Tool for making a PUT request to an API endpoint."""
name = "requests_put"
description = """Use this when you want to PUT to a website.
Input should be a json string with two keys: "url" and "data".
The value of "url" should be a string, and the value of "data" should be a dictionary of
key-value pairs you want to PUT to the url.
Be careful to always use double quotes for strings in the json string.
The output will be the text response of the PUT request.
"""
def _run(self, text: str) -> str:
"""Run the tool."""
try:
data = _parse_input(text)
return self.requests_wrapper.put(data["url"], data["data"])
except Exception as e:
return repr(e)
async def _arun(self, text: str) -> str:
"""Run the tool asynchronously."""
try:
data = _parse_input(text)
return await self.requests_wrapper.aput(data["url"], data["data"])
except Exception as e:
return repr(e)
class RequestsDeleteTool(BaseRequestsTool, BaseTool):
"""Tool for making a DELETE request to an API endpoint."""
name = "requests_delete"
description = "A portal to the internet. Use this when you need to make a DELETE request to a URL. Input should be a specific url, and the output will be the text response of the DELETE request."
def _run(self, url: str) -> str:
"""Run the tool."""
return self.requests_wrapper.delete(url)
async def _arun(self, url: str) -> str:
"""Run the tool asynchronously."""
return await self.requests_wrapper.adelete(url)
|