PricingLoginStart free
AI Agents · Leçon

Construire un pipeline d’automatisation multi-applications

Enchaîner Gmail → Slack → Google Sheets grâce à des appels d’outils gérés par un agent.

Leçon 4 sur 413 étapes

Construire un pipeline d’automatisation multi-applications est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

La chaîne de traitement multi-applications

Une chaîne de traitement multi-applications relie plusieurs services au sein d’un flux de travail automatisé unique. Dans cette leçon, nous construisons le flux suivant : nouvel e-mail Gmail → l’agent lit l’e-mail et extrait les éléments d’action → crée une carte Trello → envoie une notification Slack.

Architecture de la chaîne de traitement

La chaîne de traitement comporte quatre étapes :

  • Déclenchement : une notification push Gmail ou une interrogation périodique détecte un nouvel e-mail
  • Extraction : le LLM lit l’e-mail et extrait les éléments d’action
  • Création : l’API Trello crée une carte pour chaque élément d’action
  • Notification : l’API Slack publie un message récapitulatif

Chaque étape est une fonction distincte avec des entrées et des sorties clairement définies.

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)

Étape 1 : lecture des e-mails

Utilisez l’API Gmail pour récupérer les nouveaux e-mails. La bibliothèque google-api-python-client gère l’authentification et les appels d’API. Nous interrogeons le service à la recherche des messages qui n’ont pas encore été vus.

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 ''

Étape 2 : extraction des éléments d’action

Transmettez le contenu de l’e-mail à un LLM et demandez-lui d’extraire les éléments d’action au format JSON structuré. Utilisez response_format pour obtenir une sortie 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
    ]

Étape 3 : création de cartes Trello

L’API REST Trello crée des cartes au moyen d’une simple requête POST. Vous avez besoin de votre clé d’API, de votre jeton et de l’ID de la liste dans laquelle les cartes doivent être créées.

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_ids

Étape 4 : envoi d’une notification Slack

Utilisez le SDK Slack pour envoyer un message récapitulatif mis en forme à un canal. Le message doit résumer clairement les éléments d’action qui ont été extraits et créés.

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']

Gestion des erreurs à chaque étape

Une chaîne de traitement n’est aussi robuste que sa gestion des erreurs. Chaque étape peut échouer indépendamment. Entourez chaque appel d’une gestion des erreurs, consignez l’erreur et décidez de poursuivre ou d’interrompre la chaîne de traitement.

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
    )

Journalisation structurée des chaînes de traitement

Utilisez une journalisation structurée afin de pouvoir interroger l’historique d’exécution de la chaîne de traitement. Consignez le démarrage de la chaîne de traitement, la fin de chaque étape et le résultat final sous forme d’objets 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})

Exécuteur principal de la chaîne de traitement

Le programme principal orchestre l’ensemble. Il recherche les nouveaux e-mails, traite chacun d’eux via le pipeline, puis les marque comme lus pour éviter de les traiter à nouveau.

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)

Gestion de la configuration

Conservez tous les identifiants d’API et les paramètres du pipeline dans des variables d’environnement, et non dans le code. Chargez-les au démarrage et vérifiez que les clés obligatoires sont présentes.

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')

Tester le pipeline

Testez chaque étape du pipeline indépendamment avec des données simulées avant d’exécuter le traitement de bout en bout. Vous pouvez ainsi vérifier la logique sans consommer de crédits d’API ni créer de véritables cartes 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()

Vérification des connaissances : pipeline multi-application

Testez votre compréhension de la création de pipelines d’automatisation multi-application.

Réponds-y dans la leçon interactive

Récapitulatif du pipeline

Vous avez créé un pipeline complet d’automatisation multi-application : détection dans Gmail, extraction fondée sur un LLM, création de cartes Trello et notification Slack. Les principes de conception essentiels sont les suivants : séparer les étapes au moyen d’interfaces claires, gérer les erreurs de manière robuste à chaque étape, assurer une journalisation structurée et configurer l’application via des variables d’environnement.

Gratuit pour commencer

Apprends AI Agents avec un tuteur IA — gratuit

Écris et exécute du vrai code dans ton navigateur, obtiens de l'aide instantanée d'un tuteur IA disponible 24h/24, et reprends là où tu t'es arrêté sur le web ou dans l'app.

Cours
60
Leçons
239
Commencer gratuitementVoir le cours complet

Questions Fréquemment Posées

La leçon « Construire un pipeline d’automatisation multi-applications » est-elle gratuite ?

Oui — le texte complet de « Construire un pipeline d’automatisation multi-applications » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Construire un pipeline d’automatisation multi-applications » ?

Enchaîner Gmail → Slack → Google Sheets grâce à des appels d’outils gérés par un agent. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Construire un pipeline d’automatisation multi-applications » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Schémas d’agents déclencheur-action
  2. Connecter des agents à des webhooks
  3. Agents planifiés et fondés sur Cron
  4. Construire un pipeline d’automatisation multi-applications
← Retour à AI Agents