AI Agents · Lesson

Building a Multi-App Automation Pipeline

Chaining Gmail → Slack → Google Sheets via agent-managed tool calls.

Building a Multi-App Automation Pipeline is a free AI Agents lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

The Multi-App Pipeline

A multi-app pipeline connects several services into a single automated workflow. In this lesson we build: Gmail new email → agent reads and extracts action items → creates Trello card → sends Slack notification.

Pipeline Architecture

The pipeline has four stages:

Trigger: Gmail push notification or polling detects new email
Extract: LLM reads email and extracts action items
Create: Trello API creates a card for each action item
Notify: Slack API posts a summary message

Each stage is a separate function with clear inputs and outputs.

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)

Stage 1: Reading Emails

Use the Gmail API to fetch new emails. The google-api-python-client library handles authentication and API calls. We poll for messages not seen yet.

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

Stage 2: Extracting Action Items

Pass the email content to an LLM and ask it to extract action items in structured JSON format. Use response_format to get reliable JSON output.

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
    ]

Stage 3: Creating Trello Cards

The Trello REST API creates cards using a simple POST request. You need your API key, token, and the list ID where cards should be created.

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

Stage 4: Sending Slack Notification

Use the Slack SDK to send a formatted summary message to a channel. The message should clearly summarize what action items were extracted and created.

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

Error Handling in Each Stage

A pipeline is only as robust as its error handling. Each stage can fail independently. Wrap each call, log the error, and decide whether to continue or abort the 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
    )

Structured Logging for Pipelines

Use structured logging so you can query pipeline execution history. Log pipeline start, each stage completion, and final result as JSON objects.

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

Main Pipeline Runner

The main runner ties everything together. It polls for new emails, processes each one through the pipeline, and marks them as read to avoid reprocessing.

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)

Configuration Management

Keep all API credentials and pipeline settings in environment variables, not in code. Load them at startup and validate that required keys are present.

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

Testing the Pipeline

Test each pipeline stage independently with mock data before running end-to-end. This lets you verify logic without consuming API credits or creating real Trello cards.

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

Knowledge Check: Multi-App Pipeline

Test your understanding of building multi-app automation pipelines.

Pipeline Recap

You have built a complete multi-app automation pipeline: Gmail detection, LLM-based extraction, Trello card creation, and Slack notification. The key design principles are: separate stages with clear interfaces, robust error handling at each step, structured logging, and configuration via environment variables.

Frequently asked questions

Is the “Building a Multi-App Automation Pipeline” lesson free?

Yes — the full text of “Building a Multi-App Automation Pipeline” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.

What will I learn in “Building a Multi-App Automation Pipeline”?

Chaining Gmail → Slack → Google Sheets via agent-managed tool calls. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Agents?

No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Building a Multi-App Automation Pipeline” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Agents lesson?

Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

Trigger-Action Agent Patterns
Connecting Agents to Webhooks
Scheduling and Cron-Based Agents
Building a Multi-App Automation Pipeline

← Back to AI Agents