Creación de un agente asistente de correo sencillo
De principio a fin: leer la bandeja de entrada → resumir → redactar una respuesta → esperar aprobación.
Creación de un agente asistente de correo sencillo es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
Arquitectura de un agente asistente de correo electrónico
Un agente asistente de correo electrónico sigue una canalización estándar: Obtener → Clasificar → Decidir → Redactar → Aprobar → Enviar. El agente lee los correos no leídos, utiliza un LLM para clasificarlos y redactar respuestas, y después espera la aprobación humana antes de enviarlas. Este diseño con supervisión humana evita errores costosos derivados del envío de correos completamente autónomo.
# Email Assistant Pipeline:
#
# 1. FETCH: Pull unread emails from Gmail API
# 2. CLASSIFY: LLM labels each email
# - 'action_needed': requires a reply
# - 'fyi': informational, no reply needed
# - 'spam': should be archived
# 3. DRAFT: LLM generates reply for 'action_needed' emails
# 4. APPROVE: Human reviews drafts in Gmail UI
# 5. SEND: Agent sends approved drafts
#
# Tools: Gmail API, Anthropic/OpenAI API, json, base64
print('Email assistant pipeline: fetch -> classify -> draft -> approve -> send')Definir las herramientas del agente
estructure el agente con definiciones claras de herramientas. Cada herramienta es una función de Python con una responsabilidad específica. Esta modularidad hace que el agente sea fácil de probar, depurar y ampliar con nuevas capacidades.
class EmailAssistantTools:
def __init__(self, gmail_service, llm_client):
self.gmail = gmail_service
self.llm = llm_client
def fetch_unread(self, max_emails=10):
'''Fetch unread emails from inbox.'''
pass
def classify_email(self, subject, body, sender):
'''Ask LLM to classify: action_needed / fyi / spam.'''
pass
def draft_reply(self, subject, body, sender, context):
'''Ask LLM to draft a reply to an email.'''
pass
def create_draft(self, message_id, reply_text):
'''Save draft reply in Gmail for human review.'''
pass
def send_approved_drafts(self):
'''Send all drafts marked as approved.'''
pass
print('Tool-based architecture enables testing each step independently')Paso 1: Obtener y analizar correos electrónicos
La primera herramienta obtiene los correos no leídos y extrae los campos clave que necesita el LLM: asunto, remitente y cuerpo en texto plano. Mantenga el cuerpo breve: recórtelo a los primeros 2000 caracteres para respetar los límites de contexto del LLM y reducir los costes de la API.
import base64
def fetch_emails_for_classification(gmail_service, max_emails=10):
messages_list = gmail_service.users().messages().list(
userId='me',
q='is:unread label:inbox',
maxResults=max_emails
).execute().get('messages', [])
emails = []
for ref in messages_list:
msg = gmail_service.users().messages().get(
userId='me', id=ref['id'], format='full'
).execute()
headers = {h['name'].lower(): h['value']
for h in msg['payload'].get('headers', [])}
body = extract_plain_text_body(msg)[:2000] # trim for LLM
emails.append({
'id': msg['id'],
'thread_id': msg['threadId'],
'from': headers.get('from', ''),
'subject': headers.get('subject', '(no subject)'),
'body': body
})
print(f'Fetched {len(emails)} unread emails')
return emailsPaso 2: Clasificación mediante LLM
Envíe cada correo electrónico a un LLM para clasificarlo. Utilice un prompt estructurado que solicite al modelo un objeto JSON con la categoría y el razonamiento. Solicite explícitamente una salida JSON: es más fiable que analizar texto libre.
import json
import anthropic
import os
client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])
def classify_email(subject, sender, body):
prompt = (
'Classify this email as exactly one of: action_needed, fyi, spam.\n'
'Return JSON only: {"category": "...", "reason": "..."}\n\n'
f'From: {sender}\n'
f'Subject: {subject}\n\n'
f'Body:\n{body[:1500]}'
)
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=200,
messages=[{'role': 'user', 'content': prompt}]
)
try:
result = json.loads(response.content[0].text)
return result.get('category', 'fyi'), result.get('reason', '')
except json.JSONDecodeError:
return 'fyi', 'Could not parse LLM response'Paso 3: Generar borradores de respuesta
Para los correos clasificados como action_needed, solicite al LLM que genere un borrador de respuesta. Proporcione contexto sobre la función y el tono del agente. Solicite una respuesta profesional, pero concisa, e incluya un marcador de posición para cualquier dato que la persona deba completar.
def draft_reply(subject, sender, body, agent_context):
prompt = (
'You are an email assistant drafting a professional reply.\n'
'Guidelines:\n'
'- Be concise and professional\n'
'- Use [FILL IN] for any info you don\'t know\n'
'- Start with a greeting, end with a sign-off\n\n'
f'Context about the recipient\'s work: {agent_context}\n\n'
f'Original email from {sender}:\n'
f'Subject: {subject}\n\n'
f'{body[:1500]}\n\n'
'Draft a reply:'
)
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=500,
messages=[{'role': 'user', 'content': prompt}]
)
return response.content[0].text.strip()Paso 4: Guardar borradores para revisión humana
No envíe nunca correos generados por IA sin revisión humana. Guárdelos como borradores de Gmail para que la persona pueda abrir Gmail, revisarlos, editarlos si es necesario y enviarlos. La API para crear borradores es la misma que la API de envío; solo debe utilizar drafts().create().
import base64
from email.mime.text import MIMEText
def save_draft_reply(gmail_service, original_message, reply_text):
sender_header = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'from'), ''
)
subject = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'subject'), ''
)
msg_id_header = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'message-id'), ''
)
mime_msg = MIMEText(reply_text, 'plain', 'utf-8')
mime_msg['To'] = sender_header
mime_msg['Subject'] = 'Re: ' + subject
mime_msg['In-Reply-To'] = msg_id_header
mime_msg['References'] = msg_id_header
raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode()
draft = gmail_service.users().drafts().create(
userId='me',
body={'message': {'raw': raw, 'threadId': original_message['threadId']}}
).execute()
print(f'Draft saved: {draft["id"]}')
return draft['id']
# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
def __init__(self, result):
self._result = result
def execute(self):
return self._result
class _FakeDrafts:
def create(self, userId, body):
return _Exec({'id': 'r9000abc'})
class _FakeUsers:
def drafts(self):
return _FakeDrafts()
class _FakeGmailService:
def users(self):
return _FakeUsers()
original_message = {
'threadId': 'thread_1',
'payload': {'headers': [
{'name': 'From', 'value': 'customer@example.com'},
{'name': 'Subject', 'value': 'Question about my order'},
{'name': 'Message-ID', 'value': '<abc123@mail.example.com>'}
]}
}
save_draft_reply(_FakeGmailService(), original_message, 'Thanks for reaching out, we will look into it.')
Paso 5: Marcar los correos procesados
Después de procesar un correo (clasificarlo y crear un borrador o archivarlo), márquelo para evitar volver a procesarlo. Añada una etiqueta personalizada como AgentProcessed y quite la etiqueta UNREAD. Cree la etiqueta una sola vez si no existe.
def get_or_create_label(gmail_service, label_name):
labels = gmail_service.users().labels().list(userId='me').execute()
for label in labels.get('labels', []):
if label['name'] == label_name:
return label['id']
# Create the label
new_label = gmail_service.users().labels().create(
userId='me',
body={
'name': label_name,
'labelListVisibility': 'labelShow',
'messageListVisibility': 'show'
}
).execute()
print(f'Created label: {label_name}')
return new_label['id']
def mark_processed(gmail_service, message_id, agent_label_id):
gmail_service.users().messages().modify(
userId='me',
id=message_id,
body={
'addLabelIds': [agent_label_id],
'removeLabelIds': ['UNREAD']
}
).execute()
# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
def __init__(self, result):
self._result = result
def execute(self):
return self._result
class _FakeUsers:
def __init__(self):
self._labels = [{'id': 'Label_1', 'name': 'Processed'}]
def labels(self):
return self
def list(self, userId):
return _Exec({'labels': self._labels})
def create(self, userId, body):
print(f'Created label: {body["name"]}')
return _Exec({'id': 'Label_2', 'name': body['name']})
def messages(self):
return self
def modify(self, userId, id, body):
print(f'Marked {id} processed with {body}')
return _Exec({'id': id})
class _FakeGmailService:
def users(self):
return _FakeUsers()
gmail_service = _FakeGmailService()
label_id = get_or_create_label(gmail_service, 'Processed')
print(f'Label id: {label_id}')
new_label_id = get_or_create_label(gmail_service, 'AgentHandled')
mark_processed(gmail_service, 'msg_99', new_label_id)
Bucle completo de ejecución del agente
Conecte todos los pasos en un único método run(). El agente recorre los correos, clasifica cada uno, crea borradores para los que requieren una acción, archiva el spam y marca todos como procesados. Registre un resumen al final.
def run_email_agent(gmail_service, agent_context, max_emails=10):
processed_label = get_or_create_label(gmail_service, 'AgentProcessed')
emails = fetch_emails_for_classification(gmail_service, max_emails)
summary = {'action_needed': 0, 'fyi': 0, 'spam': 0, 'drafts_created': 0}
for email in emails:
category, reason = classify_email(
email['subject'], email['from'], email['body']
)
summary[category] += 1
print(f'[{category}] {email["subject"][:60]} - {reason[:50]}')
if category == 'action_needed':
# Fetch full message for reply context
full_msg = gmail_service.users().messages().get(
userId='me', id=email['id'], format='full'
).execute()
reply = draft_reply(
email['subject'], email['from'],
email['body'], agent_context
)
save_draft_reply(gmail_service, full_msg, reply)
summary['drafts_created'] += 1
mark_processed(gmail_service, email['id'], processed_label)
print('\nAgent run complete:')
for k, v in summary.items():
print(f' {k}: {v}')
return summaryGestión de errores y resiliencia
Encapsule cada paso del procesamiento de correos en un bloque try/except. Un error en un correo nunca debe impedir que el agente procese los demás. Registre los errores con el ID y el asunto del correo para poder revisarlos más adelante.
from googleapiclient.errors import HttpError
def process_email_safely(gmail_service, email, agent_context, label_id):
try:
category, reason = classify_email(
email['subject'], email['from'], email['body']
)
if category == 'action_needed':
full_msg = gmail_service.users().messages().get(
userId='me', id=email['id'], format='full'
).execute()
reply = draft_reply(
email['subject'], email['from'],
email['body'], agent_context
)
save_draft_reply(gmail_service, full_msg, reply)
mark_processed(gmail_service, email['id'], label_id)
return category
except HttpError as e:
print(f'Gmail API error on {email["id"]}: {e.resp.status}')
return 'error'
except Exception as e:
print(f'Unexpected error on {email["subject"][:50]}: {e}')
return 'error'Programar la ejecución periódica del agente
Ejecute el agente de correo según un horario mediante un bucle sencillo con un time.sleep() entre ejecuciones. En producción, utilice un programador de tareas como cron, APScheduler o una función en la nube. Añada siempre una comprobación para evitar ejecuciones fuera del horario laboral y respetar las zonas horarias de los destinatarios.
import time
import datetime
def is_business_hours():
now = datetime.datetime.now()
# Mon-Fri, 9am-6pm local time
return (now.weekday() < 5 and 9 <= now.hour < 18)
def run_scheduler(gmail_service, agent_context,
interval_minutes=30, max_emails=20):
print(f'Email agent started. Checking every {interval_minutes} min.')
while True:
if is_business_hours():
print(f'\n[{datetime.datetime.now().strftime("%H:%M")}] Running agent...')
try:
run_email_agent(gmail_service, agent_context, max_emails)
except Exception as e:
print(f'Agent run failed: {e}')
else:
print('Outside business hours, skipping run')
time.sleep(interval_minutes * 60)Registro y seguimiento de auditoría
Mantenga un registro JSON de cada correo que procese el agente. Esto crea un seguimiento de auditoría que puede revisar para verificar el comportamiento del agente, diagnosticar clasificaciones incorrectas y mejorar los prompts con el tiempo.
import json
import datetime
from pathlib import Path
LOG_FILE = Path('agent_audit.jsonl')
def log_action(email_id, subject, sender, category, action, draft_id=None):
entry = {
'timestamp': datetime.datetime.now().isoformat(),
'email_id': email_id,
'subject': subject[:100],
'from': sender,
'category': category,
'action': action,
'draft_id': draft_id
}
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(json.dumps(entry) + '\n')
# Usage in the main loop:
log_action(
email_id='18abc123',
subject='Invoice #1234',
sender='billing@vendor.com',
category='action_needed',
action='draft_created',
draft_id='r9000abc'
)
# --- demo ---
print('Audit log contents:')
print(LOG_FILE.read_text(encoding='utf-8'))
Comprobación rápida: supervisión humana
Compruebe sus conocimientos sobre el diseño del asistente de correo electrónico.
Repaso del agente asistente de correo electrónico
Ha creado una canalización completa para un agente asistente de correo electrónico:
- Obtener: recuperar correos no leídos mediante consultas de búsqueda de la API de Gmail
- Clasificar: el LLM clasifica cada correo como action_needed/fyi/spam con una salida JSON
- Redactar: el LLM genera una respuesta con tono profesional y marcadores [FILL IN]
- Guardar el borrador: almacenarlo en Gmail para revisión humana; nunca enviarlo automáticamente
- Marcar como procesado: añadir una etiqueta personalizada y quitar UNREAD para evitar volver a procesarlo
- Aislar errores: envolver cada correo en try/except para evitar que un error detenga el lote
- Registro de auditoría: archivo JSONL que registra cada acción para su revisión y mejora
Preguntas frecuentes
¿La lección «Creación de un agente asistente de correo sencillo» es gratis?
Sí — el texto completo de «Creación de un agente asistente de correo sencillo» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Creación de un agente asistente de correo sencillo»?
De principio a fin: leer la bandeja de entrada → resumir → redactar una respuesta → esperar aprobación. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.
¿Cuánto tiempo toma la lección «Creación de un agente asistente de correo sencillo»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Conexión con Gmail mediante la API
- Lectura y envío programático de correos electrónicos
- Creación y consulta de eventos de calendario
- Creación de un agente asistente de correo sencillo