"""
Email Agent Orchestrator
Main agent that coordinates all email processing tasks
"""
import os
import time
from typing import Optional, List
from datetime import datetime
from email_classifier import EmailClassifier, EmailMessage, ClassificationResult
from gmail_integration import GmailClient
from sendgrid_integration import SendGridClient
from database import EmailDatabase
from response_templates import ResponseTemplates
from cold_outreach_db import ColdOutreachDB
import logging
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 = 'email_agent.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()