Creación de un pipeline de automatización multiaplicación
Encadenamiento de Gmail → Slack → Google Sheets mediante llamadas a herramientas gestionadas por agentes.
Creación de un pipeline de automatización multiaplicación 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.
El pipeline entre varias aplicaciones
Un pipeline entre varias aplicaciones conecta varios servicios en un único flujo de trabajo automatizado. En esta lección crearemos lo siguiente: nuevo correo electrónico de Gmail → el agente lo lee y extrae las tareas pendientes → crea una tarjeta de Trello → envía una notificación de Slack.
Arquitectura del pipeline
El pipeline tiene cuatro etapas:
- Disparador: una notificación push de Gmail o el polling detecta un correo electrónico nuevo
- Extracción: el LLM lee el correo electrónico y extrae las tareas pendientes
- Creación: la API de Trello crea una tarjeta por cada tarea pendiente
- Notificación: la API de Slack publica un mensaje resumido
Cada etapa es una función independiente con entradas y salidas claras.
from dataclasses import dataclass, field
from typing import List
@dataclass
class Email:
id: str
sender: str
subject: str
body: str
@dataclass
class ActionItem:
title: str
description: str
due_date: str = None
@dataclass
class PipelineResult:
email_id: str
action_items: List[ActionItem] = field(default_factory=list)
trello_card_ids: List[str] = field(default_factory=list)
slack_message_ts: str = None
error: str = None
if __name__ == '__main__':
email = Email(id='e1', sender='alice@example.com', subject='Project update', body='See attached.')
result = PipelineResult(email_id=email.id, action_items=[ActionItem(title='Review attachment', description='Check the doc')])
print(f'Pipeline result for {result.email_id}: {len(result.action_items)} action item(s)')
print(' -', result.action_items[0].title)
Etapa 1: lectura de correos electrónicos
Utilice la API de Gmail para obtener correos electrónicos nuevos. La biblioteca google-api-python-client gestiona la autenticación y las llamadas a la API. Realizamos polling para buscar mensajes que aún no se hayan visto.
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import base64
def get_gmail_service(token_path='token.json'):
creds = Credentials.from_authorized_user_file(token_path)
return build('gmail', 'v1', credentials=creds)
def fetch_unread_emails(service, max_results=10):
results = service.users().messages().list(
userId='me',
q='is:unread',
maxResults=max_results
).execute()
messages = results.get('messages', [])
emails = []
for msg in messages:
detail = service.users().messages().get(
userId='me', id=msg['id'], format='full'
).execute()
headers = {h['name']: h['value'] for h in detail['payload']['headers']}
emails.append(Email(
id=msg['id'],
sender=headers.get('From', ''),
subject=headers.get('Subject', ''),
body=extract_body(detail)
))
return emails
def extract_body(message_detail):
payload = message_detail.get('payload', {})
if 'data' in payload.get('body', {}):
return base64.urlsafe_b64decode(payload['body']['data']).decode('utf-8')
return ''Etapa 2: extracción de tareas pendientes
Pase el contenido del correo electrónico a un LLM y solicítele que extraiga las tareas pendientes en formato JSON estructurado. Utilice response_format para obtener una salida JSON fiable.
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def extract_action_items(email: 'Email') -> list:
prompt = (
'Extract all action items from this email. '
'Return JSON array with objects having fields: '
'title (string), description (string), due_date (string or null).\n\n'
f'From: {email.sender}\n'
f'Subject: {email.subject}\n'
f'Body:\n{email.body}'
)
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
response_format={'type': 'json_object'}
)
result = json.loads(response.choices[0].message.content)
items = result.get('action_items', [])
return [
ActionItem(
title=item['title'],
description=item.get('description', ''),
due_date=item.get('due_date')
)
for item in items
]Etapa 3: creación de tarjetas de Trello
La API REST de Trello crea tarjetas mediante una sencilla solicitud POST. Necesita su clave de API, su token y el ID de la lista en la que deben crearse las tarjetas.
import httpx
TRELLO_API_KEY = 'your-trello-api-key'
TRELLO_TOKEN = 'your-trello-token'
TRELLO_LIST_ID = 'your-list-id'
def create_trello_card(action_item: 'ActionItem') -> str:
url = 'https://api.trello.com/1/cards'
params = {
'key': TRELLO_API_KEY,
'token': TRELLO_TOKEN
}
data = {
'name': action_item.title,
'desc': action_item.description,
'idList': TRELLO_LIST_ID,
'due': action_item.due_date
}
response = httpx.post(url, params=params, json=data)
response.raise_for_status()
card = response.json()
return card['id']
def create_trello_cards_for_items(action_items: list) -> list:
card_ids = []
for item in action_items:
card_id = create_trello_card(item)
print(f'Created Trello card: {item.title} (ID: {card_id})')
card_ids.append(card_id)
return card_idsEtapa 4: envío de una notificación de Slack
Utilice el SDK de Slack para enviar un mensaje resumido con formato a un canal. El mensaje debe resumir claramente qué tareas pendientes se extrajeron y se crearon.
from slack_sdk import WebClient
slack_client = WebClient(token='xoxb-your-slack-bot-token')
def send_slack_summary(email: 'Email', action_items: list, card_ids: list, channel: str = '#automation'):
if not action_items:
return None
items_text = '\n'.join([
f' - {item.title}'
for item in action_items
])
message = (
f'*New email processed from {email.sender}*\n'
f'*Subject:* {email.subject}\n\n'
f'*Action items extracted ({len(action_items)}):*\n'
f'{items_text}\n\n'
f'Trello cards created: {len(card_ids)}'
)
response = slack_client.chat_postMessage(
channel=channel,
text=message,
mrkdwn=True
)
return response['ts']Gestión de errores en cada etapa
La solidez de un pipeline depende de su gestión de errores. Cada etapa puede fallar de forma independiente. Envuelva cada llamada, registre el error y decida si debe continuar o cancelar el pipeline.
import logging
logger = logging.getLogger('pipeline')
def run_pipeline_stage(stage_name, fn, *args, **kwargs):
try:
result = fn(*args, **kwargs)
logger.info(f'Stage {stage_name}: success')
return result, None
except Exception as e:
logger.error(f'Stage {stage_name}: FAILED - {e}')
return None, str(e)
def process_email_pipeline(email):
# Stage 2: Extract
action_items, err = run_pipeline_stage('extract', extract_action_items, email)
if err:
return PipelineResult(email_id=email.id, error=f'Extract failed: {err}')
if not action_items:
logger.info(f'No action items found in email {email.id}')
return PipelineResult(email_id=email.id, action_items=[])
# Stage 3: Trello
card_ids, err = run_pipeline_stage('trello', create_trello_cards_for_items, action_items)
if err:
card_ids = [] # Continue even if Trello fails
# Stage 4: Slack
ts, err = run_pipeline_stage('slack', send_slack_summary, email, action_items, card_ids or [])
return PipelineResult(
email_id=email.id,
action_items=action_items,
trello_card_ids=card_ids or [],
slack_message_ts=ts
)Registro estructurado para pipelines
Utilice el registro estructurado para poder consultar el historial de ejecuciones del pipeline. Registre el inicio del pipeline, la finalización de cada etapa y el resultado final como objetos JSON.
import logging
import json
from datetime import datetime
class PipelineLogger:
def __init__(self, pipeline_name):
self.pipeline_name = pipeline_name
self.logger = logging.getLogger(pipeline_name)
self.run_id = None
self.start_time = None
def start(self, email_id):
self.run_id = f'{email_id}_{int(datetime.now().timestamp())}'
self.start_time = datetime.now()
self.logger.info(json.dumps({
'event': 'pipeline_start',
'run_id': self.run_id,
'email_id': email_id
}))
def stage_done(self, stage, result_summary):
self.logger.info(json.dumps({
'event': 'stage_complete',
'run_id': self.run_id,
'stage': stage,
'result': result_summary
}))
def finish(self, success, details):
duration = (datetime.now() - self.start_time).total_seconds()
self.logger.info(json.dumps({
'event': 'pipeline_finish',
'run_id': self.run_id,
'success': success,
'duration_seconds': duration,
'details': details
}))
if __name__ == '__main__':
import sys
logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stdout)
pl = PipelineLogger('demo_pipeline')
pl.start('email_123')
pl.stage_done('extract', {'items_found': 3})
pl.finish(True, {'action_items': 2})
Ejecutor principal del pipeline
El ejecutor principal lo coordina todo. Consulta periódicamente si hay correos nuevos, procesa cada uno mediante el pipeline y los marca como leídos para evitar volver a procesarlos.
import time
def mark_as_read(gmail_service, email_id):
gmail_service.users().messages().modify(
userId='me',
id=email_id,
body={'removeLabelIds': ['UNREAD']}
).execute()
def run_email_pipeline_loop(gmail_service, poll_interval=60):
pipeline_logger = PipelineLogger('email_pipeline')
print(f'Pipeline running. Polling every {poll_interval}s')
while True:
try:
emails = fetch_unread_emails(gmail_service)
print(f'Found {len(emails)} unread emails')
for email in emails:
pipeline_logger.start(email.id)
result = process_email_pipeline(email)
if result.error:
pipeline_logger.finish(False, {'error': result.error})
else:
mark_as_read(gmail_service, email.id)
pipeline_logger.finish(True, {
'action_items': len(result.action_items),
'cards_created': len(result.trello_card_ids)
})
except Exception as e:
print(f'Pipeline loop error: {e}')
time.sleep(poll_interval)Gestión de la configuración
Mantenga todas las credenciales de API y la configuración del pipeline en variables de entorno, no en el código. Cárguelas al iniciar la aplicación y valide que estén presentes las claves obligatorias.
import os
from dotenv import load_dotenv
load_dotenv()
class PipelineConfig:
def __init__(self):
self.openai_api_key = os.environ.get('OPENAI_API_KEY', '')
self.slack_bot_token = os.environ.get('SLACK_BOT_TOKEN', '')
self.trello_api_key = os.environ.get('TRELLO_API_KEY', '')
self.trello_token = os.environ.get('TRELLO_TOKEN', '')
self.trello_list_id = os.environ.get('TRELLO_LIST_ID', '')
self.slack_channel = os.environ.get('SLACK_CHANNEL', '#automation')
self.poll_interval = int(os.environ.get('POLL_INTERVAL_SECONDS', '60'))
def validate(self):
missing = []
required = [
('OPENAI_API_KEY', self.openai_api_key),
('SLACK_BOT_TOKEN', self.slack_bot_token),
('TRELLO_API_KEY', self.trello_api_key),
('TRELLO_TOKEN', self.trello_token),
('TRELLO_LIST_ID', self.trello_list_id)
]
for name, value in required:
if not value:
missing.append(name)
if missing:
raise ValueError(f'Missing required config: {missing}')
return True
config = PipelineConfig()
config.validate()
print('Config validated successfully')Pruebas del pipeline
Pruebe cada etapa del pipeline de forma independiente con datos simulados antes de ejecutarlo de extremo a extremo. Así podrá verificar la lógica sin consumir créditos de API ni crear tarjetas reales en Trello.
from unittest.mock import MagicMock, patch
def test_extract_action_items_mock():
mock_response = MagicMock()
mock_response.choices[0].message.content = '{"action_items": [{"title": "Follow up with vendor", "description": "Call about invoice", "due_date": null}]}'
with patch('openai.OpenAI') as mock_openai:
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = mock_response
mock_openai.return_value = mock_client
test_email = Email(
id='test123',
sender='vendor@example.com',
subject='Invoice Follow-up Needed',
body='Please follow up with the vendor about the outstanding invoice.'
)
# Would call extract_action_items(test_email) with mocked OpenAI
print('Test email:', test_email.subject)
print('Mock response parsed successfully')
test_extract_action_items_mock()Comprobación de conocimientos: pipeline multiaplicación
Compruebe su comprensión sobre la creación de pipelines de automatización para varias aplicaciones.
Resumen del pipeline
Ha creado un pipeline completo de automatización para varias aplicaciones: detección en Gmail, extracción basada en un LLM, creación de tarjetas en Trello y notificación en Slack. Los principios clave de diseño son: separar las etapas mediante interfaces claras, gestionar los errores de forma sólida en cada paso, usar registros estructurados y configurar el sistema mediante variables de entorno.
Preguntas frecuentes
¿La lección «Creación de un pipeline de automatización multiaplicación» es gratis?
Sí — el texto completo de «Creación de un pipeline de automatización multiaplicación» 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 pipeline de automatización multiaplicación»?
Encadenamiento de Gmail → Slack → Google Sheets mediante llamadas a herramientas gestionadas por agentes. 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 pipeline de automatización multiaplicación»?
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
- Patrones de agentes de tipo activador-acción
- Conexión de agentes a webhooks
- Agentes programados y basados en cron
- Creación de un pipeline de automatización multiaplicación