import os import requests from datetime import datetime, timedelta from database import notifications_collection, users_collection, agents_collection def is_mailgun_configured() -> bool: """Check if Mailgun environment variables are set.""" return bool(os.getenv("MAILGUN_API_KEY")) and bool(os.getenv("MAILGUN_DOMAIN")) def send_mailgun_email(to_emails: list[str], subject: str, html_body: str) -> bool: """Send email via Mailgun HTTP API. Returns True on success.""" api_key = os.getenv("MAILGUN_API_KEY") domain = os.getenv("MAILGUN_DOMAIN") from_email = os.getenv("MAILGUN_FROM_EMAIL", f"AgentHub ") response = requests.post( f"https://api.mailgun.net/v3/{domain}/messages", auth=("api", api_key), data={ "from": from_email, "to": to_emails, "subject": subject, "html": html_body, }, timeout=10, ) return response.status_code == 200 def build_threshold_email(agent_name: str, weekly_tokens: int, threshold: int) -> str: """Build HTML email body for a weekly token threshold notification.""" return f"""

AgentHub - High Weekly Token Usage Alert

The following agent has exceeded the weekly token usage threshold:

Agent Name	{agent_name}
Tokens (Last 7 Days)	{weekly_tokens:,}
Weekly Threshold	{threshold:,}

This is an automated notification from AgentHub. Please review the agent's recent token consumption.

""" async def check_and_notify_threshold(agent_name: str): """Check 7-day token usage against threshold and send notification if exceeded. Non-blocking, safe to call.""" if not is_mailgun_configured(): return threshold = int(os.getenv("TOKEN_USAGE_THRESHOLD", "100000")) # Calculate token usage over the last 7 days from usage_timeline agent = await agents_collection.find_one( {"agent_name": agent_name}, {"usage_timeline": 1} ) if not agent or not agent.get("usage_timeline"): return cutoff_date = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d") weekly_tokens = sum( entry.get("token_count", 0) for entry in agent["usage_timeline"] if entry.get("date", "") >= cutoff_date ) if weekly_tokens < threshold: return cooldown_hours = int(os.getenv("NOTIFICATION_COOLDOWN_HOURS", "24")) cooldown_cutoff = datetime.utcnow() - timedelta(hours=cooldown_hours) # Check cooldown - skip if we already notified within the cooldown period recent = await notifications_collection.find_one({ "agent_name": agent_name, "sent_at": {"$gte": cooldown_cutoff}, }) if recent: return # Get admin user emails (exclude placeholder accounts with no real domain) admin_cursor = users_collection.find( {"is_admin": True, "is_active": True, "email": {"$not": {"$regex": r"@agenthub\.com$"}}}, {"email": 1}, ) admin_emails = [doc["email"] async for doc in admin_cursor] if not admin_emails: return # Send email subject = f"[AgentHub] High Weekly Token Usage: {agent_name} ({weekly_tokens:,} tokens in 7 days)" html_body = build_threshold_email(agent_name, weekly_tokens, threshold) success = send_mailgun_email(admin_emails, subject, html_body) # Record the notification attempt await notifications_collection.insert_one({ "agent_name": agent_name, "weekly_tokens": weekly_tokens, "threshold": threshold, "recipients": admin_emails, "success": success, "sent_at": datetime.utcnow(), }) def build_client_agent_email(agent_data: dict) -> str: """Build HTML email body for a client agent creation notification.""" return f"""

Client Agent Created

A new client-facing agent has been created and requires verification.

Agent Name	{agent_data.get('agent_name', 'N/A')}
Description	{agent_data.get('agent_description', 'N/A')}
Purpose	{agent_data.get('agent_purpose', 'N/A')}
Client	Yes
Client Name	{agent_data.get('client_name', 'N/A')}
Studio Name	{agent_data.get('studio_name', 'N/A')}
Tool	{agent_data.get('agent_tool', 'N/A')}
Created By	{agent_data.get('created_by_email', 'N/A')}

Please review this agent in AgentHub.

""" def send_client_agent_notification(agent_data: dict): """Send email notification when a client-facing agent is created. Non-blocking.""" if not is_mailgun_configured(): return notify_emails_str = os.getenv("CLIENT_AGENT_NOTIFY_EMAILS", "") if not notify_emails_str: return to_emails = [e.strip() for e in notify_emails_str.split(",") if e.strip()] if not to_emails: return try: subject = "Client Agent Created" html_body = build_client_agent_email(agent_data) send_mailgun_email(to_emails, subject, html_body) except Exception as e: print(f"Failed to send client agent notification: {e}") def build_weekly_digest_email(agents: list) -> str: """Build HTML email body for the weekly agent digest.""" rows = "" for i, agent in enumerate(agents, 1): rows += f""" {i}. {agent.get('agent_name', 'N/A')}
Purpose: {agent.get('agent_purpose', 'N/A')}
Description: {agent.get('agent_description', 'N/A')}
Created by: {agent.get('created_by_email', 'N/A')} """ return f"""

Agents Created in Last Week

{len(agents)} agent(s) created in the last week:

{rows}

View full details in the AgentHub admin dashboard.

""" async def send_weekly_agent_digest(): """Send weekly digest of agents created in the last 7 days to all admins.""" if not is_mailgun_configured(): print("Weekly digest: Mailgun not configured, skipping.") return cutoff = datetime.utcnow() - timedelta(days=7) # Find agents created in last 7 days cursor = agents_collection.find( {"created_at": {"$gte": cutoff}}, {"agent_name": 1, "agent_purpose": 1, "agent_description": 1, "created_by": 1} ).sort("created_at", -1) agents = await cursor.to_list(length=None) if not agents: print("Weekly digest: No agents created in last 7 days, skipping.") return # Resolve creator emails for agent in agents: created_by = agent.get("created_by", "") if created_by == "agent_collector_api": agent["created_by_email"] = "Agent Collector API" else: try: from bson import ObjectId user = await users_collection.find_one({"_id": ObjectId(created_by)}, {"email": 1}) agent["created_by_email"] = user["email"] if user else created_by except Exception: agent["created_by_email"] = created_by # Get admin emails (exclude placeholder accounts with no real domain) admin_cursor = users_collection.find( {"is_admin": True, "is_active": True, "email": {"$not": {"$regex": r"@agenthub\.com$"}}}, {"email": 1}, ) admin_emails = [doc["email"] async for doc in admin_cursor] if not admin_emails: print("Weekly digest: No admin emails found, skipping.") return subject = "Agents Created in Last Week" html_body = build_weekly_digest_email(agents) try: success = send_mailgun_email(admin_emails, subject, html_body) print(f"Weekly digest: Sent to {len(admin_emails)} admins, success={success}") except Exception as e: print(f"Weekly digest: Failed to send: {e}")