Spaces:
Runtime error
Runtime error
File size: 2,617 Bytes
7a158ef f4a4e0a 7a158ef f4a4e0a 7a158ef |
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 |
import os
from dotenv import load_dotenv
from agency_swarm.tools import BaseTool
from pydantic import Field
from typing import Dict, Any
load_dotenv()
# Load API credentials from environment variables
api_key = os.getenv("EVOLUTION_API_KEY")
api_url = os.getenv("EVOLUTION_API_URL")
api_instance = os.getenv("EVOLUTION_API_INSTANCE")
class SendWhatsAppText(BaseTool):
"""
Tool for sending WhatsApp text messages to users.
This tool sends a text message to a specified phone number using the Evolution API.
Use this when you need to request authorization, feedback, or additional information
from a human user to make progress on a task or project.
"""
# Add example_field with a default value to satisfy BaseTool validation
example_field: str = Field(
default="whatsapp_text",
description="Identifier for this tool. Can be left at its default value.",
)
phone_number: str = Field(
...,
description="Recipient's phone number in E.164 format (e.g., '5521988456100').",
)
message: str = Field(..., description="The text message content to send.")
def run(self) -> Dict[str, Any]:
"""
Send a WhatsApp text message to the specified phone number.
Returns:
dict: The JSON response from the API containing the message ID on success,
or error details on failure.
"""
import requests
# Build the complete URL
instance_name = os.getenv("EVOLUTION_API_INSTANCE", "Cogmo_Secretary_Joao")
url = f"{os.getenv('EVOLUTION_API_URL')}/message/sendText/{instance_name}"
# Set up the headers with API key authentication
headers = {
"Content-Type": "application/json",
"apikey": os.getenv("EVOLUTION_API_KEY"),
}
text_message = (
"Agency Swarm Project Manager: \n"
+ self.message
+ "\n\n"
+ "--------------------------------\n"
+ "This message was sent by an AI agent. *Please do not reply*"
)
# Prepare the request body
data = {"number": self.phone_number, "text": text_message}
try:
# Make the POST request
response = requests.post(url, headers=headers, json=data)
# Return the JSON response
return response.json()
except Exception as e:
# Handle any exceptions
return {
"error": str(e),
"status": "failed",
"message": "Failed to send WhatsApp message",
}
|