127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
import os
|
|
import requests
|
|
import threading
|
|
import time
|
|
from langchain_core.tools import tool
|
|
from langchain_google_genai import ChatGoogleGenerativeAI
|
|
from langgraph.prebuilt import create_react_agent
|
|
from temporalio import activity
|
|
|
|
WHATSAPP_GATEWAY_URL = "http://whatsapp-gateway.ai-agents.svc.cluster.local:5001/api/send-message"
|
|
|
|
@tool
|
|
def send_whatsapp(target_jid: str, message: str) -> str:
|
|
"""Send a WhatsApp message to a specific group or user (target_jid)."""
|
|
try:
|
|
resp = requests.post(WHATSAPP_GATEWAY_URL, json={"to": target_jid, "message": message})
|
|
print(f"Sent WhatsApp: {resp.json()}")
|
|
return f"WhatsApp sent successfully to {target_jid}"
|
|
except Exception as e:
|
|
print(f"Failed to send WhatsApp: {e}")
|
|
return f"Failed to send: {str(e)}"
|
|
|
|
def _delayed_reminder(target_jid: str, message: str, delay_seconds: int):
|
|
time.sleep(delay_seconds)
|
|
try:
|
|
requests.post(WHATSAPP_GATEWAY_URL, json={"to": target_jid, "message": f"⏰ *REMINDER*\n\n{message}"})
|
|
except Exception as e:
|
|
print(f"Reminder delivery failed: {e}")
|
|
|
|
@tool
|
|
def set_reminder(target_jid: str, reminder_text: str, delay_seconds: int) -> str:
|
|
"""Set a reminder that will send a WhatsApp message back to the target_jid after delay_seconds.
|
|
You must convert the user's requested time into seconds for the delay_seconds argument.
|
|
"""
|
|
thread = threading.Thread(target=_delayed_reminder, args=(target_jid, reminder_text, delay_seconds))
|
|
thread.daemon = True
|
|
thread.start()
|
|
return f"Reminder scheduled for {delay_seconds} seconds from now."
|
|
|
|
@tool
|
|
def schedule_calendar(event: str) -> str:
|
|
"""Schedule a calendar event."""
|
|
print(f"Mock Calendar scheduled: {event}")
|
|
return f"Calendar event scheduled: {event}"
|
|
|
|
@tool
|
|
def trigger_utility_scraper() -> str:
|
|
"""Trigger the utility scraper to run and collect utility data."""
|
|
url = "http://utility-agent.ai-agents.svc.cluster.local:5005/api/run"
|
|
try:
|
|
resp = requests.post(url, timeout=15)
|
|
resp.raise_for_status()
|
|
return f"Utility scraper triggered successfully. Response: {resp.text}"
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Failed to trigger utility scraper: {e}")
|
|
return f"Failed to trigger utility scraper: {str(e)}"
|
|
|
|
from kubernetes import client, config
|
|
|
|
@tool
|
|
def get_k8s_pods(namespace: str) -> str:
|
|
"""Get the status of all pods in a specific Kubernetes namespace."""
|
|
try:
|
|
config.load_incluster_config()
|
|
v1 = client.CoreV1Api()
|
|
pods = v1.list_namespaced_pod(namespace)
|
|
result = f"Pods in {namespace}:\n"
|
|
for p in pods.items:
|
|
result += f"- {p.metadata.name} (Status: {p.status.phase})\n"
|
|
return result
|
|
except Exception as e:
|
|
return f"K8s Error: {str(e)}"
|
|
|
|
@tool
|
|
def delete_k8s_pod(pod_name: str, namespace: str) -> str:
|
|
"""Delete a specific pod in a Kubernetes namespace to restart it."""
|
|
try:
|
|
config.load_incluster_config()
|
|
v1 = client.CoreV1Api()
|
|
v1.delete_namespaced_pod(name=pod_name, namespace=namespace)
|
|
return f"Successfully deleted pod {pod_name} in {namespace}. It will automatically restart."
|
|
except Exception as e:
|
|
return f"K8s Error: {str(e)}"
|
|
|
|
GMAIL_AGENT_URL = "http://gmail-agent.ai-agents.svc.cluster.local:5002/api/send-email"
|
|
|
|
@tool
|
|
def send_email(to_email: str, subject: str, body: str) -> str:
|
|
"""Send an email response back to the user."""
|
|
try:
|
|
resp = requests.post(GMAIL_AGENT_URL, json={"to": to_email, "subject": subject, "body": body})
|
|
print(f"Sent Email: {resp.json()}")
|
|
return f"Email sent successfully to {to_email}"
|
|
except Exception as e:
|
|
print(f"Failed to send Email: {e}")
|
|
return f"Failed to send email: {str(e)}"
|
|
|
|
tools = [send_whatsapp, set_reminder, schedule_calendar, trigger_utility_scraper, get_k8s_pods, delete_k8s_pod, send_email]
|
|
|
|
@activity.defn
|
|
def run_agent_activity(payload: dict) -> str:
|
|
user_request = payload.get("text", "")
|
|
target_jid = payload.get("targetJid", "default")
|
|
sender = payload.get("sender", "Unknown")
|
|
|
|
llm = ChatGoogleGenerativeAI(model="gemini-1.5-pro")
|
|
|
|
system_prompt = f"""You are a highly capable AI assistant operating as an orchestrator.
|
|
|
|
The user ({sender}) has sent a message.
|
|
Their routing ID (target_jid) is: {target_jid}.
|
|
|
|
ROUTING INSTRUCTIONS:
|
|
- If target_jid is 'EMAIL', it means the user sent this request via Email. You MUST use the `send_email` tool to reply to them (set `to_email` to {sender}, and choose a relevant `subject`).
|
|
- Otherwise, the user sent this request via WhatsApp. You MUST use the `send_whatsapp` tool to reply and pass {target_jid} as the target_jid!
|
|
|
|
TOOLS AVAILABLE:
|
|
- Scrape utility data: call the `trigger_utility_scraper` tool.
|
|
- Reminders: use the `set_reminder` tool (pass {target_jid} as target_jid unless it's EMAIL, then fallback to WhatsApp or don't set reminders for email).
|
|
- Check Kubernetes pods: use the `get_k8s_pods` tool.
|
|
- Restart a pod or deployment: use the `delete_k8s_pod` tool.
|
|
"""
|
|
|
|
agent = create_react_agent(llm, tools)
|
|
result = agent.invoke({"messages": [("system", system_prompt), ("user", user_request)]})
|
|
return result["messages"][-1].content
|