AI Agents · Lección

Lectura y envío programático de correos electrónicos

Enumere mensajes, obtenga el cuerpo de los mensajes y envíe correos MIME.

Lectura y envío programático de correos electrónicos es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 2 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.

Listado de mensajes con la Gmail API

El primer paso para leer el correo es listar los mensajes que coincidan con sus criterios. service.users().messages().list() devuelve los ID de los mensajes y los ID de los hilos, no el contenido completo. Después, recupere cada mensaje individualmente. Este patrón de dos pasos mantiene rápida la llamada de listado.

def list_messages(service, user_id='me', query='', max_results=10):
    results = service.users().messages().list(
        userId=user_id,
        q=query,          # Gmail search query
        maxResults=max_results
    ).execute()

    messages = results.get('messages', [])
    print(f'Found {len(messages)} messages')
    return messages

# Examples of Gmail search queries:
# 'is:unread' — unread messages
# 'from:boss@company.com is:unread' — unread from boss
# 'subject:invoice label:inbox' — invoices in inbox
# 'after:2026/05/01 has:attachment' — recent with attachments
messages = list_messages(gmail_service, query='is:unread label:inbox')

Obtención de un mensaje completo

Use service.users().messages().get() para recuperar todo el contenido del mensaje. El parámetro format controla la cantidad de datos que se devuelve: 'full' incluye los encabezados y el cuerpo, 'metadata' devuelve únicamente los encabezados y 'minimal' devuelve solo los ID y las etiquetas.

def get_message(service, message_id, user_id='me'):
    message = service.users().messages().get(
        userId=user_id,
        id=message_id,
        format='full'  # 'full', 'metadata', or 'minimal'
    ).execute()
    return message

# Fetch the first unread message
messages = list_messages(gmail_service, query='is:unread', max_results=1)
if messages:
    msg = get_message(gmail_service, messages[0]['id'])
    print('Thread ID:', msg['threadId'])
    print('Labels:', msg['labelIds'])
    print('Snippet:', msg['snippet'][:100])

Extracción de encabezados de correo electrónico

Los encabezados (From, To, Subject, Date) se almacenan en message['payload']['headers'] como una lista de diccionarios {'name': ..., 'value': ...}. Escriba una función auxiliar para extraer encabezados por nombre; la usará constantemente.

def get_header(message, name):
    headers = message.get('payload', {}).get('headers', [])
    for h in headers:
        if h['name'].lower() == name.lower():
            return h['value']
    return ''

def extract_email_meta(message):
    return {
        'id': message['id'],
        'from': get_header(message, 'From'),
        'to': get_header(message, 'To'),
        'subject': get_header(message, 'Subject'),
        'date': get_header(message, 'Date'),
        'snippet': message.get('snippet', '')
    }

meta = extract_email_meta(msg)
print(f'From: {meta["from"]}')
print(f'Subject: {meta["subject"]}')
print(f'Date: {meta["date"]}')

Decodificación del cuerpo del correo electrónico (base64)

Los cuerpos de los correos electrónicos en la Gmail API están codificados con base64url, una variante de base64 segura para URL en la que + se convierte en - y / se convierte en _. Use base64.urlsafe_b64decode() para decodificarlos. Gestione tanto los correos simples (de una sola parte) como los multiparte.

import base64

def decode_body(data):
    if not data:
        return ''
    decoded_bytes = base64.urlsafe_b64decode(data + '==')
    return decoded_bytes.decode('utf-8', errors='replace')

def get_email_body(message):
    payload = message.get('payload', {})
    mime_type = payload.get('mimeType', '')

    # Simple (non-multipart) email
    if 'body' in payload and payload['body'].get('data'):
        return decode_body(payload['body']['data'])

    # Multipart email: find the text/plain or text/html part
    parts = payload.get('parts', [])
    for part in parts:
        if part.get('mimeType') == 'text/plain':
            return decode_body(part['body'].get('data', ''))

    # Fallback: try text/html
    for part in parts:
        if part.get('mimeType') == 'text/html':
            return decode_body(part['body'].get('data', ''))

    return message.get('snippet', '')

# --- demo ---
encoded = base64.urlsafe_b64encode(b'Hello from the agent!').decode().rstrip('=')
print('Decoded body:', decode_body(encoded))

message = {
    'payload': {
        'mimeType': 'multipart/alternative',
        'parts': [
            {'mimeType': 'text/plain', 'body': {'data': encoded}}
        ]
    }
}
print('Email body:', get_email_body(message))

Gestión recursiva de correos multiparte

Los correos complejos (con archivos adjuntos, imágenes insertadas o contenido mixto) son estructuras multiparte anidadas. Las partes del cuerpo pueden anidarse a cualquier profundidad. Una función recursiva que recorra el árbol de partes gestiona todos los casos.

import base64

def extract_parts(payload, target_mime='text/plain'):
    parts_text = []
    mime_type = payload.get('mimeType', '')

    if mime_type == target_mime:
        data = payload.get('body', {}).get('data', '')
        if data:
            decoded = base64.urlsafe_b64decode(data + '==').decode('utf-8', errors='replace')
            parts_text.append(decoded)

    # Recurse into sub-parts
    for part in payload.get('parts', []):
        parts_text.extend(extract_parts(part, target_mime))

    return parts_text

def get_plain_text(message):
    payload = message.get('payload', {})
    texts = extract_parts(payload, 'text/plain')
    return '\n\n'.join(texts) if texts else message.get('snippet', '')

body_text = get_plain_text(msg)
print(f'Body ({len(body_text)} chars):', body_text[:200])

Marcar mensajes como leídos

Después de procesar un correo, el agente debe marcarlo como leído eliminando la etiqueta UNREAD. Use service.users().messages().modify() con removeLabelIds=['UNREAD']. También puede añadir etiquetas como PROCESSED para realizar un seguimiento de los correos gestionados por el agente.

def mark_as_read(service, message_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'removeLabelIds': ['UNREAD']}
    ).execute()
    print(f'Marked {message_id} as read')

def add_label(service, message_id, label_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'addLabelIds': [label_id]}
    ).execute()

# Get label ID by name
def get_label_id(service, label_name, user_id='me'):
    labels = service.users().labels().list(userId=user_id).execute()
    for label in labels.get('labels', []):
        if label['name'].lower() == label_name.lower():
            return label['id']
    return None

# --- 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 messages(self):
        return self
    def modify(self, **kwargs):
        print(f'[gmail api] messages.modify({kwargs})')
        return _Exec({'id': kwargs.get('id')})
    def labels(self):
        return self
    def list(self, **kwargs):
        return _Exec({'labels': [{'id': 'Label_1', 'name': 'Processed'}]})

class _FakeService:
    def users(self):
        return _FakeUsers()

service = _FakeService()
mark_as_read(service, 'msg_42')
add_label(service, 'msg_42', 'Label_1')
print('Label id for "Processed":', get_label_id(service, 'Processed'))

Redacción de correos con MIMEText

Para enviar un correo electrónico, primero redáctelo como un mensaje MIME mediante la biblioteca estándar email de Python. Después, codifique en base64url los bytes sin procesar y envíelos mediante una solicitud POST a la Gmail API. MIMEText gestiona la codificación correcta del cuerpo del mensaje.

import base64
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def create_message(sender, to, subject, body_text, body_html=None):
    if body_html:
        msg = MIMEMultipart('alternative')
        msg.attach(MIMEText(body_text, 'plain', 'utf-8'))
        msg.attach(MIMEText(body_html, 'html', 'utf-8'))
    else:
        msg = MIMEText(body_text, 'plain', 'utf-8')

    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject

    # Encode as base64url
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {'raw': raw}

message = create_message(
    sender='agent@yourcompany.com',
    to='recipient@example.com',
    subject='Weekly Summary',
    body_text='Hello,\n\nHere is your summary.\n\nBest,\nAgent'
)

# --- demo ---
print('Message keys:', list(message.keys()))
print('Base64 length:', len(message['raw']))

Envío de correo con la Gmail API

Envíe el mensaje redactado mediante service.users().messages().send(). El parámetro userId='me' hace referencia al usuario autenticado. La API devuelve el mensaje enviado junto con su ID y el ID del hilo.

from googleapiclient.errors import HttpError

def send_message(service, message, user_id='me'):
    try:
        sent = service.users().messages().send(
            userId=user_id,
            body=message
        ).execute()
        print(f'Message sent! ID: {sent["id"]}')
        return sent
    except HttpError as e:
        import json
        body = json.loads(e.content.decode())
        print(f'Send failed ({e.resp.status}): {body.get("error", {}).get("message")}')
        return None

# Send the message
message = create_message(
    sender='me',
    to='team@company.com',
    subject='Agent Report',
    body_text='Processing complete. 42 tasks handled.'
)
send_message(gmail_service, message)

Creación y envío de borradores de correo

En lugar de enviarlos inmediatamente, los agentes pueden crear borradores para que los revise una persona. Use service.users().drafts().create(). Después, una persona puede revisar y enviar el borrador desde la interfaz de Gmail. Este es el patrón recomendado para cualquier correo que requiera aprobación humana.

def create_draft(service, message, user_id='me'):
    draft = service.users().drafts().create(
        userId=user_id,
        body={'message': message}
    ).execute()
    print(f'Draft created: {draft["id"]}')
    return draft

def send_draft(service, draft_id, user_id='me'):
    sent = service.users().drafts().send(
        userId=user_id,
        body={'id': draft_id}
    ).execute()
    print(f'Draft sent as message: {sent["id"]}')
    return sent

# Create a draft for review
message = create_message(
    sender='me',
    to='client@example.com',
    subject='Proposal Follow-up',
    body_text='Dear Client,\n\nFollowing up on our proposal...'
)
draft = create_draft(gmail_service, message)
# Human reviews in Gmail, then agent sends:
# send_draft(gmail_service, draft['id'])

Procesamiento por lotes de varios correos

Cuando procese muchos correos, no los obtenga uno por uno en un bucle sin pausas, ya que alcanzará los límites de cuota. Use un bucle controlado con pequeñas pausas o la función de solicitudes por lotes de la Gmail API para agrupar varias operaciones en una sola llamada HTTP.

import time

def process_unread_emails(service, max_emails=20):
    messages = list_messages(
        service,
        query='is:unread label:inbox',
        max_results=max_emails
    )

    processed = []
    for i, msg_ref in enumerate(messages):
        # Rate-limit: process max 5 per second
        if i > 0 and i % 5 == 0:
            time.sleep(1)

        msg = get_message(service, msg_ref['id'])
        meta = extract_email_meta(msg)
        body = get_plain_text(msg)

        result = {
            'id': msg['id'],
            'from': meta['from'],
            'subject': meta['subject'],
            'body_preview': body[:200]
        }
        processed.append(result)
        mark_as_read(service, msg['id'])

    return processed

Respuesta a un correo (en el mismo hilo)

Para enviar una respuesta en el mismo hilo, establezca los encabezados In-Reply-To y References en el encabezado Message-ID del mensaje original y pase el threadId a la llamada de envío. De este modo, la respuesta se mantiene en el mismo hilo de conversación de Gmail.

import base64
from email.mime.text import MIMEText

def create_reply(original_message, reply_text, sender='me'):
    original_msg_id = get_header(original_message, 'Message-ID')
    to = get_header(original_message, 'From')
    subject = get_header(original_message, 'Subject')
    if not subject.startswith('Re:'):
        subject = 'Re: ' + subject

    msg = MIMEText(reply_text, 'plain', 'utf-8')
    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject
    msg['In-Reply-To'] = original_msg_id
    msg['References'] = original_msg_id

    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {
        'raw': raw,
        'threadId': original_message['threadId']  # keeps it in thread
    }

Comprobación rápida: cuerpo de correo en base64

Compruebe su comprensión de la gestión de mensajes de la Gmail API.

Resumen de lectura y envío de correos

Su agente ya puede leer y enviar correo mediante programación:

Listar: messages().list(q='is:unread') devuelve los ID; use la sintaxis de búsqueda de Gmail
Obtener: messages().get(id=..., format='full') devuelve el mensaje completo
Analizar encabezados: extraiga From/Subject/Date de payload.headers
Decodificar el cuerpo: use base64.urlsafe_b64decode(data) para el texto; recorra recursivamente las partes multiparte
Enviar: redacte con MIMEText, codifique en base64url y envíe mediante messages().send()
Responder en el hilo: establezca el encabezado In-Reply-To y threadId en el cuerpo del envío

Preguntas frecuentes

¿La lección «Lectura y envío programático de correos electrónicos» es gratis?

Sí — el texto completo de «Lectura y envío programático de correos electrónicos» 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 «Lectura y envío programático de correos electrónicos»?

Enumere mensajes, obtenga el cuerpo de los mensajes y envíe correos MIME. 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 2 de 4.

¿Cuánto tiempo toma la lección «Lectura y envío programático de correos electrónicos»?

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

← Volver a AI Agents