Community: Infobip tool integration

docs/docs/integrations/tools/infobip.ipynb

@@ -0,0 +1,176 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Infobip\n",
+    "This notebook that shows how to use [Infobip](https://www.infobip.com/) API wrapper to send SMS messages, emails.\n",
+    "\n",
+    "Infobip provides many services, but this notebook will focus on SMS and Email services. You can find more information about the API and other channels [here](https://www.infobip.com/docs/api)."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Setup\n",
+    "\n",
+    "To use this tool you need to have an Infobip account. You can create [free trial account](https://www.infobip.com/docs/essentials/free-trial).\n",
+    "\n",
+    "\n",
+    "`InfobipAPIWrapper` uses name parameters where you can provide credentials:\n",
+    "\n",
+    "- `infobip_api_key` - [API Key](https://www.infobip.com/docs/essentials/api-authentication#api-key-header) that you can find in your [developer tools](https://portal.infobip.com/dev/api-keys)\n",
+    "- `infobip_base_url` - [Base url](https://www.infobip.com/docs/essentials/base-url) for Infobip API. You can use default value `https://api.infobip.com/`.\n",
+    "\n",
+    "You can also provide `infobip_api_key` and `infobip_base_url` as environment variables `INFOBIP_API_KEY` and `INFOBIP_BASE_URL`."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Sending a SMS"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "vscode": {
+     "languageId": "plaintext"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "from langchain_community.utilities.infobip import InfobipAPIWrapper\n",
+    "\n",
+    "infobip: InfobipAPIWrapper = InfobipAPIWrapper()\n",
+    "\n",
+    "infobip.run(\n",
+    "    to=\"41793026727\",\n",
+    "    text=\"Hello, World!\",\n",
+    "    sender=\"Langchain\",\n",
+    "    channel=\"sms\",\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Sending a Email"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "vscode": {
+     "languageId": "plaintext"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "from langchain_community.utilities.infobip import InfobipAPIWrapper\n",
+    "\n",
+    "infobip: InfobipAPIWrapper = InfobipAPIWrapper()\n",
+    "\n",
+    "infobip.run(\n",
+    "    to=\"[email protected]\",\n",
+    "    sender=\"[email protected]\",\n",
+    "    subject=\"example\",\n",
+    "    body=\"example\",\n",
+    "    channel=\"email\",\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# How to use it inside an Agent "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "vscode": {
+     "languageId": "plaintext"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "from langchain import hub\n",
+    "from langchain.agents import AgentExecutor, create_openai_functions_agent\n",
+    "from langchain.tools import StructuredTool\n",
+    "from langchain_community.utilities.infobip import InfobipAPIWrapper\n",
+    "from langchain_core.pydantic_v1 import BaseModel, Field\n",
+    "from langchain_openai import ChatOpenAI\n",
+    "\n",
+    "instructions = \"You are a coding teacher. You are teaching a student how to code. The student asks you a question. You answer the question.\"\n",
+    "base_prompt = hub.pull(\"langchain-ai/openai-functions-template\")\n",
+    "prompt = base_prompt.partial(instructions=instructions)\n",
+    "llm = ChatOpenAI(temperature=0)\n",
+    "\n",
+    "\n",
+    "class EmailInput(BaseModel):\n",
+    "    body: str = Field(description=\"Email body text\")\n",
+    "    to: str = Field(description=\"Email address to send to. Example: [email protected]\")\n",
+    "    sender: str = Field(\n",
+    "        description=\"Email address to send from, must be '[email protected]'\"\n",
+    "    )\n",
+    "    subject: str = Field(description=\"Email subject\")\n",
+    "    channel: str = Field(description=\"Email channel, must be 'email'\")\n",
+    "\n",
+    "\n",
+    "infobip_api_wrapper: InfobipAPIWrapper = InfobipAPIWrapper()\n",
+    "infobip_tool = StructuredTool.from_function(\n",
+    "    name=\"infobip_email\",\n",
+    "    description=\"Send Email via Infobip. If you need to send email, use infobip_email\",\n",
+    "    func=infobip_api_wrapper.run,\n",
+    "    args_schema=EmailInput,\n",
+    ")\n",
+    "tools = [infobip_tool]\n",
+    "\n",
+    "agent = create_openai_functions_agent(llm, tools, prompt)\n",
+    "agent_executor = AgentExecutor(\n",
+    "    agent=agent,\n",
+    "    tools=tools,\n",
+    "    verbose=True,\n",
+    ")\n",
+    "\n",
+    "agent_executor.invoke(\n",
+    "    {\n",
+    "        \"input\": \"Hi, can you please send me an example of Python recursion to my email [email protected]\"\n",
+    "    }\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "```bash\n",
+    "> Entering new AgentExecutor chain...\n",
+    "\n",
+    "Invoking: `infobip_email` with `{'body': 'Hi,\\n\\nHere is a simple example of a recursive function in Python:\\n\\n```\\ndef factorial(n):\\n    if n == 1:\\n        return 1\\n    else:\\n        return n * factorial(n-1)\\n```\\n\\nThis function calculates the factorial of a number. The factorial of a number is the product of all positive integers less than or equal to that number. The function calls itself with a smaller argument until it reaches the base case where n equals 1.\\n\\nBest,\\nCoding Teacher', 'to': '[email protected]', 'sender': '[email protected]', 'subject': 'Python Recursion Example', 'channel': 'email'}`\n",
+    "\n",
+    "\n",
+    "I have sent an example of Python recursion to your email. Please check your inbox.\n",
+    "\n",
+    "> Finished chain.\n",
+    "```"
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

libs/community/langchain_community/utilities/__init__.py

@@ -128,6 +128,12 @@ def _import_graphql() -> Any:
     return GraphQLAPIWrapper
 
 
+def _import_infobip() -> Any:
+    from langchain_community.utilities.infobip import InfobipAPIWrapper
+
+    return InfobipAPIWrapper
+
+
 def _import_jira() -> Any:
     from langchain_community.utilities.jira import JiraAPIWrapper
 
@@ -335,6 +341,8 @@ def __getattr__(name: str) -> Any:
         return _import_google_serper()
     elif name == "GraphQLAPIWrapper":
         return _import_graphql()
+    elif name == "InfobipAPIWrapper":
+        return _import_infobip()
     elif name == "JiraAPIWrapper":
         return _import_jira()
     elif name == "MaxComputeAPIWrapper":
@@ -414,6 +422,7 @@ def __getattr__(name: str) -> Any:
     "GoogleSearchAPIWrapper",
     "GoogleSerperAPIWrapper",
     "GraphQLAPIWrapper",
+    "InfobipAPIWrapper",
     "JiraAPIWrapper",
     "LambdaWrapper",
     "MaxComputeAPIWrapper",

libs/community/langchain_community/utilities/infobip.py

@@ -0,0 +1,178 @@
+"""Util that sends messages via Infobip."""
+from typing import Dict, List, Optional
+
+import requests
+from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator
+from langchain_core.utils import get_from_dict_or_env
+from requests.adapters import HTTPAdapter
+from requests_toolbelt import MultipartEncoder
+from urllib3.util import Retry
+
+
+class InfobipAPIWrapper(BaseModel):
+    """Wrapper for Infobip API for messaging."""
+
+    infobip_api_key: Optional[str] = None
+    infobip_base_url: Optional[str] = "https://api.infobip.com"
+
+    class Config:
+        """Configuration for this pydantic object."""
+
+        extra = Extra.forbid
+
+    @root_validator(pre=True)
+    def validate_environment(cls, values: Dict) -> Dict:
+        """Validate that api key exists in environment."""
+        values["infobip_api_key"] = get_from_dict_or_env(
+            values, "infobip_api_key", "INFOBIP_API_KEY"
+        )
+        values["infobip_base_url"] = get_from_dict_or_env(
+            values, "infobip_base_url", "INFOBIP_BASE_URL"
+        )
+        return values
+
+    def _get_requests_session(self) -> requests.Session:
+        """Get a requests session with the correct headers."""
+        retry_strategy: Retry = Retry(
+            total=4,  # Maximum number of retries
+            backoff_factor=2,  # Exponential backoff factor
+            status_forcelist=[429, 500, 502, 503, 504],  # HTTP status codes to retry on
+        )
+        adapter: HTTPAdapter = HTTPAdapter(max_retries=retry_strategy)
+
+        session = requests.Session()
+        session.mount("https://", adapter)
+        session.headers.update(
+            {
+                "Authorization": f"App {self.infobip_api_key}",
+                "User-Agent": "infobip-langchain-community",
+            }
+        )
+        return session
+
+    def _send_sms(
+        self, sender: str, destination_phone_numbers: List[str], text: str
+    ) -> str:
+        """Send an SMS message."""
+        json: Dict = {
+            "messages": [
+                {
+                    "destinations": [
+                        {"to": destination} for destination in destination_phone_numbers
+                    ],
+                    "from": sender,
+                    "text": text,
+                }
+            ]
+        }
+
+        session: requests.Session = self._get_requests_session()
+        session.headers.update(
+            {
+                "Content-Type": "application/json",
+            }
+        )
+
+        response: requests.Response = session.post(
+            f"{self.infobip_base_url}/sms/2/text/advanced",
+            json=json,
+        )
+
+        response_json: Dict = response.json()
+        try:
+            if response.status_code != 200:
+                return response_json["requestError"]["serviceException"]["text"]
+        except KeyError:
+            return "Failed to send message"
+
+        try:
+            return response_json["messages"][0]["messageId"]
+        except KeyError:
+            return (
+                "Could not get message ID from response, message was sent successfully"
+            )
+
+    def _send_email(
+        self, from_email: str, to_email: str, subject: str, body: str
+    ) -> str:
+        """Send an email message."""
+        form_data: Dict = {
+            "from": from_email,
+            "to": to_email,
+            "subject": subject,
+            "text": body,
+        }
+
+        data = MultipartEncoder(fields=form_data)
+
+        session: requests.Session = self._get_requests_session()
+        session.headers.update(
+            {
+                "Content-Type": data.content_type,
+            }
+        )
+
+        response: requests.Response = session.post(
+            f"{self.infobip_base_url}/email/3/send",
+            data=data,
+        )
+
+        response_json: Dict = response.json()
+
+        try:
+            if response.status_code != 200:
+                return response_json["requestError"]["serviceException"]["text"]
+        except KeyError:
+            return "Failed to send message"
+
+        try:
+            return response_json["messages"][0]["messageId"]
+        except KeyError:
+            return (
+                "Could not get message ID from response, message was sent successfully"
+            )
+
+    def run(
+        self,
+        body: str = "",
+        to: str = "",
+        sender: str = "",
+        subject: str = "",
+        channel: str = "sms",
+    ) -> str:
+        if channel == "sms":
+            if sender == "":
+                raise ValueError("Sender must be specified for SMS messages")
+
+            if to == "":
+                raise ValueError("Destination must be specified for SMS messages")
+
+            if body == "":
+                raise ValueError("Body must be specified for SMS messages")
+
+            return self._send_sms(
+                sender=sender,
+                destination_phone_numbers=[to],
+                text=body,
+            )
+        elif channel == "email":
+            if sender == "":
+                raise ValueError("Sender must be specified for email messages")
+
+            if to == "":
+                raise ValueError("Destination must be specified for email messages")
+
+            if subject == "":
+                raise ValueError("Subject must be specified for email messages")
+
+            if body == "":
+                raise ValueError("Body must be specified for email messages")
+
+            return self._send_email(
+                from_email=sender,
+                to_email=to,
+                subject=subject,
+                body=body,
+            )
+        else:
+            raise ValueError(f"Channel {channel} is not supported")