Source code for langchain.tools.requests.tool
# flake8: noqa
"""Tools for making requests to an API endpoint."""
import json
from typing import Any, Dict, Optional
from langchain.pydantic_v1 import BaseModel
from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain.utilities.requests import TextRequestsWrapper
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)
def _clean_url(url: str) -> str:
    """Strips quotes from the url."""
    return url.strip("\"'")
[docs]class BaseRequestsTool(BaseModel):
    """Base class for requests tools."""
    requests_wrapper: TextRequestsWrapper
[docs]class RequestsGetTool(BaseRequestsTool, BaseTool):
    """Tool for making a GET request to an API endpoint."""
    name: str = "requests_get"
    description: str = "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, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Run the tool."""
        return self.requests_wrapper.get(_clean_url(url))
    async def _arun(
        self,
        url: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Run the tool asynchronously."""
        return await self.requests_wrapper.aget(_clean_url(url))