"""
Email Agent Orchestrator
Main agent that coordinates all email processing tasks
"""
import sys
import os
import time
from pathlib import Path
from typing import Optional, List
from datetime import datetime
# K:\projekte-AG als Package-Root setzen
sys.path.insert(0, str(Path(__file__).parents[2]))
from agents.email_agent.email_classifier import EmailClassifier, EmailMessage, ClassificationResult
from agents.email_agent.gmail_integration import GmailClient
from agents.email_agent.sendgrid_integration import SendGridClient
from agents.email_agent.database import EmailDatabase
from agents.email_agent.response_templates import ResponseTemplates
from lead_engine.storage.cold_outreach_db import ColdOutreachDB
import logging
_DEFAULT_DB = str(Path(__file__).parents[2] / 'lead_engine' / 'storage' / 'database.db')
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class EmailAgent:
"""Main Email Agent - orchestrates the entire email processing workflow"""
def __init__(self,
credentials_file: str = 'credentials.json',
sendgrid_api_key: Optional[str] = None,
anthropic_api_key: Optional[str] = None,
db_path: str = _DEFAULT_DB):
"""
Initialize Email Agent with all dependencies
Args:
credentials_file: Gmail OAuth credentials file
sendgrid_api_key: SendGrid API key (or use env var)
anthropic_api_key: Anthropic API key (or use env var)
db_path: Database file path
"""
logger.info("Initializing Email Agent...")
self.classifier = EmailClassifier(api_key=anthropic_api_key)
self.gmail = GmailClient(credentials_file=credentials_file)
self.sendgrid = SendGridClient(api_key=sendgrid_api_key)
self.db = EmailDatabase(db_path=db_path)
self.outreach_db = ColdOutreachDB(db_path=db_path)
logger.info("Email Agent initialized successfully")
def process_incoming_emails(self, max_results: int = 10) -> dict:
"""
Process all unread emails
Args:
max_results: Maximum number of emails to process
Returns:
Dictionary with processing results
"""
logger.info(f"Processing incoming emails (max: {max_results})...")
results = {
'processed': 0,
'auto_replied': 0,
'escalated': 0,
'logged': 0,
'failed': 0,
'details': []
}
# Get unread emails from Gmail
unread_emails = self.gmail.get_unread_messages(max_results=max_results)
logger.info(f"Found {len(unread_emails)} unread emails")
for email_data in unread_emails:
try:
result = self._process_single_email(email_data)
results['processed'] += 1
if result['action'] == 'auto_reply':
results['auto_replied'] += 1
elif result['action'] == 'escalate':
results['escalated'] += 1
elif result['action'] == 'log_only':
results['logged'] += 1
results['details'].append(result)
logger.info(f"Processed email from {email_data['from_email']}: {result['action']}")
# Mark as read
self.gmail.mark_as_read(email_data['message_id'])
# Reply-Detection: War dies eine Antwort auf eine Outreach-Email?
self._check_outreach_reply(email_data)
except Exception as e:
results['failed'] += 1
logger.error(f"Error processing email: {e}")
results['details'].append({
'from_email': email_data.get('from_email'),
'error': str(e)
})
return results
def _process_single_email(self, email_data: dict) -> dict:
"""
Process a single email through classification and response
Args:
email_data: Email data from Gmail
Returns:
Dictionary with processing result
"""
# Create EmailMessage object
email_msg = EmailMessage(
from_email=email_data['from_email'],
from_name=email_data['from_name'],
subject=email_data['subject'],
body=email_data['body'],
timestamp=email_data['timestamp'],
message_id=email_data['message_id']
)
# Classify email
classification = self.classifier.classify_email(email_msg)
# Log the email
log_id = self.db.log_email({
'from_email': email_data['from_email'],
'from_name': email_data['from_name'],
'subject': email_data['subject'],
'body': email_data['body'],
'category': classification.category,
'priority': classification.priority,
'sentiment': classification.sentiment,
'action_taken': classification.suggested_action,
'message_id': email_data['message_id']
})
logger.info(f"Logged email {log_id}: {classification.category} ({classification.priority})")
# Take action based on classification
action_result = self._take_action(
classification=classification,
email_data=email_data,
log_id=log_id
)
return {
'log_id': log_id,
'from_email': email_data['from_email'],
'category': classification.category,
'priority': classification.priority,
'sentiment': classification.sentiment,
'action': classification.suggested_action,
'reasoning': classification.reasoning,
'confidence': classification.confidence,
'action_result': action_result
}
def _take_action(self, classification: ClassificationResult,
email_data: dict, log_id: int) -> dict:
"""
Take appropriate action based on email classification
Args:
classification: Classification result
email_data: Original email data
log_id: Database log ID
Returns:
Dictionary with action result
"""
action = classification.suggested_action
action_data = {}
if action == 'auto_reply':
action_data = self._auto_reply(
classification=classification,
email_data=email_data,
log_id=log_id
)
elif action == 'escalate':
action_data = self._escalate_to_human(
classification=classification,
email_data=email_data,
log_id=log_id
)
elif action == 'route_to_sales':
action_data = self._route_to_sales(
email_data=email_data,
log_id=log_id
)
elif action == 'log_only':
action_data = {'status': 'logged', 'log_id': log_id}
return action_data
def _auto_reply(self, classification: ClassificationResult,
email_data: dict, log_id: int) -> dict:
"""Send automatic reply"""
template_name = self.classifier.get_template_for_category(classification.category)
if not template_name:
return {'status': 'failed', 'reason': 'No template found'}
# Render template
template = ResponseTemplates.render_template(
category=classification.category,
context={'sender_name': email_data['from_name']}
)
if not template:
return {'status': 'failed', 'reason': 'Template render failed'}
# Queue email for sending
queue_id = self.db.queue_email_response({
'to_email': email_data['from_email'],
'to_name': email_data['from_name'],
'subject': template['subject'],
'body': template['body'],
'action': 'auto_reply'
})
logger.info(f"Queued auto-reply for {email_data['from_email']} (queue_id: {queue_id})")
return {
'status': 'queued',
'queue_id': queue_id,
'template': classification.category
}
def _escalate_to_human(self, classification: ClassificationResult,
email_data: dict, log_id: int) -> dict:
"""Escalate email to human agent"""
# Queue escalation email
template = ResponseTemplates.render_template(
'escalation',
context={'sender_name': email_data['from_name']}
)
queue_id = self.db.queue_email_response({
'to_email': email_data['from_email'],
'to_name': email_data['from_name'],
'subject': template['subject'],
'body': template['body'],
'action': 'escalate'
})
# Mark in database as escalated
self.db.update_email_log(log_id, {
'action_taken': 'escalate',
'resolved': 0
})
logger.info(f"Escalated email from {email_data['from_email']} to human agent")
return {
'status': 'escalated',
'queue_id': queue_id,
'priority': classification.priority
}
def _check_outreach_reply(self, email_data: dict):
"""
Prueft ob eine eingehende Email eine Antwort auf eine Outreach-Email ist.
Wenn ja: Kontakt-Status auf 'replied' setzen und alle pending Sequenzen stoppen.
"""
from_email = email_data.get('from_email', '').lower().strip()
if not from_email:
return
if self.outreach_db.email_exists(from_email):
self.outreach_db.mark_replied(from_email)
logger.info(
f"OUTREACH-ANTWORT erkannt von {from_email}! "
f"Sequenz gestoppt. Bitte manuell nachfassen."
)
# TODO: Notification an Karlo (z.B. per Email oder Telegram)
def _route_to_sales(self, email_data: dict, log_id: int) -> dict:
"""Route email to sales team"""
template = ResponseTemplates.render_template(
'sales_routing',
context={'sender_name': email_data['from_name']}
)
queue_id = self.db.queue_email_response({
'to_email': email_data['from_email'],
'to_name': email_data['from_name'],
'subject': template['subject'],
'body': template['body'],
'action': 'route_to_sales'
})
logger.info(f"Routed email from {email_data['from_email']} to sales")
return {
'status': 'routed',
'queue_id': queue_id,
'department': 'sales'
}
def send_queued_emails(self) -> dict:
"""Send all queued emails"""
logger.info("Sending queued emails...")
results = {'sent': 0, 'failed': 0}
pending = self.db.get_pending_emails()
for email in pending:
try:
success = self.sendgrid.send_email(
to_email=email['to_email'],
to_name=email['to_name'],
subject=email['subject'],
body=email['body']
)
if success:
self.db.mark_email_sent(email['id'])
results['sent'] += 1
logger.info(f"Sent email to {email['to_email']}")
else:
self.db.mark_email_failed(email['id'], 'SendGrid returned error')
results['failed'] += 1
logger.error(f"Failed to send email to {email['to_email']}")
except Exception as e:
self.db.mark_email_failed(email['id'], str(e))
results['failed'] += 1
logger.error(f"Exception sending email to {email['to_email']}: {e}")
return results
def get_statistics(self) -> dict:
"""Get email agent statistics"""
return self.db.get_statistics()
def run_continuous(self, check_interval: int = 60, send_interval: int = 30):
"""
Run agent continuously
Args:
check_interval: Seconds between checking for new emails
send_interval: Seconds between sending queued emails
"""
logger.info(f"Starting Email Agent (check every {check_interval}s, send every {send_interval}s)")
last_check = 0
last_send = 0
iteration = 0
try:
while True:
current_time = time.time()
iteration += 1
# Check for new emails
if current_time - last_check >= check_interval:
logger.info(f"[Iteration {iteration}] Checking for new emails...")
results = self.process_incoming_emails()
logger.info(f"Processed: {results['processed']}, "
f"Auto-replied: {results['auto_replied']}, "
f"Escalated: {results['escalated']}, "
f"Failed: {results['failed']}")
last_check = current_time
# Send queued emails
if current_time - last_send >= send_interval:
logger.info(f"[Iteration {iteration}] Sending queued emails...")
send_results = self.send_queued_emails()
logger.info(f"Sent: {send_results['sent']}, Failed: {send_results['failed']}")
last_send = current_time
time.sleep(5) # Small sleep to prevent busy-waiting
except KeyboardInterrupt:
logger.info("Email Agent stopped by user")
except Exception as e:
logger.error(f"Email Agent error: {e}")
if __name__ == '__main__':
# Initialize and run agent
agent = EmailAgent()
# Option 1: Process emails once
# results = agent.process_incoming_emails()
# print(f"Results: {results}")
# Option 2: Run continuously
agent.run_continuous()