Archives
/
langchain
mirror of https://github.com/hwchase17/langchain
2
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

community[minor]: Infobip tool integration (#16805)

Browse Source 
**Description:** Adding Tool that wraps Infobip API for sending sms or
emails and email validation.
**Dependencies:** None,
**Twitter handle:** @hmilkovic

Implementation:
```
libs/community/langchain_community/utilities/infobip.py
```

Integration tests:
```
libs/community/tests/integration_tests/utilities/test_infobip.py
```

Example notebook:
```
docs/docs/integrations/tools/infobip.ipynb
```

---------

Co-authored-by: Bagatur <baskaryan@gmail.com>
pull/18934/head
Hrvoje Milković 3 months ago committed by GitHub
parent 727a2ea9f1
commit b7344e3347
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 449 additions and 0 deletions
Show Stats Download Patch File Download Diff File

176
docs/docs/integrations/tools/infobip.ipynb
Unescape Escape View File

@ -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=\"test@example.com\",\n",
" sender=\"test@example.com\",\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@example.com\")\n",
" sender: str = Field(\n",
" description=\"Email address to send from, must be 'validemail@example.com'\"\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@example.com\"\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@example.com', 'sender': 'validemail@example.com', '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
}

1
libs/community/langchain_community/utilities/__init__.py
Unescape Escape View File

@ -26,6 +26,7 @@ _module_lookup = {
"GoogleSerperAPIWrapper": "langchain_community.utilities.google_serper",
"GoogleTrendsAPIWrapper": "langchain_community.utilities.google_trends",
"GraphQLAPIWrapper": "langchain_community.utilities.graphql",
"InfobipAPIWrapper": "langchain_community.utilities.infobip",
"JiraAPIWrapper": "langchain_community.utilities.jira",
"LambdaWrapper": "langchain_community.utilities.awslambda",
"MaxComputeAPIWrapper": "langchain_community.utilities.max_compute",

185
libs/community/langchain_community/utilities/infobip.py
Unescape Escape View File

@ -0,0 +1,185 @@
"""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 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."""
try:
from requests_toolbelt import MultipartEncoder
except ImportError as e:
raise ImportError(
"Unable to import requests_toolbelt, please install it with "
"`pip install -U requests-toolbelt`."
) from e
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")

86
libs/community/tests/integration_tests/utilities/test_infobip.py
Unescape Escape View File

@ -0,0 +1,86 @@
from typing import Dict
import responses
from langchain_community.utilities.infobip import InfobipAPIWrapper
def test_send_sms() -> None:
infobip: InfobipAPIWrapper = InfobipAPIWrapper(
infobip_api_key="test",
infobip_base_url="https://api.infobip.com",
)
json_response: Dict = {
"messages": [
{
"messageId": "123",
"status": {
"description": "Message sent to next instance",
"groupId": 1,
"groupName": "PENDING",
"id": 26,
"name": "PENDING_ACCEPTED",
},
"to": "41793026727",
}
]
}
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://api.infobip.com/sms/2/text/advanced",
json=json_response,
status=200,
)
response: str = infobip.run(
body="test",
to="41793026727",
sender="41793026727",
channel="sms",
)
assert response == "123"
def test_send_email() -> None:
infobip: InfobipAPIWrapper = InfobipAPIWrapper(
infobip_api_key="test",
infobip_base_url="https://api.infobip.com",
)
json_response: Dict = {
"bulkId": "123",
"messages": [
{
"to": "test@example.com",
"messageId": "123",
"status": {
"groupId": 1,
"groupName": "PENDING",
"id": 26,
"name": "PENDING_ACCEPTED",
"description": "Message accepted, pending for delivery.",
},
}
],
}
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://api.infobip.com/email/3/send",
json=json_response,
status=200,
)
response: str = infobip.run(
body="test",
to="test@example.com",
sender="test@example.com",
subject="test",
channel="email",
)
assert response == "123"

1
libs/community/tests/unit_tests/utilities/test_imports.py
Unescape Escape View File

@ -19,6 +19,7 @@ EXPECTED_ALL = [
"GoogleSerperAPIWrapper",
"GoogleTrendsAPIWrapper",
"GraphQLAPIWrapper",
"InfobipAPIWrapper",
"JiraAPIWrapper",
"LambdaWrapper",
"MaxComputeAPIWrapper",

Write Preview
Loading…
Cancel
Save