0Pricing
AI Agents · レッスン

複数アプリの自動化パイプラインを構築する

エージェントが管理するツール呼び出しで Gmail → Slack → Google Sheets を連携します。

「複数アプリの自動化パイプラインを構築する」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

マルチアプリパイプライン

マルチアプリパイプラインは、複数のサービスを1つの自動化されたワークフローに接続します。このレッスンでは、Gmailの新着メール → エージェントが読み取ってアクションアイテムを抽出 → Trelloカードを作成 → Slack通知を送信という流れを構築します。

パイプラインのアーキテクチャ

パイプラインは4つの段階で構成されます。

  • Trigger：Gmailのプッシュ通知またはポーリングで新しいメールを検知します
  • Extract：LLMがメールを読み取り、アクションアイテムを抽出します
  • Create：Trello APIが各アクションアイテムのカードを作成します
  • Notify：Slack APIがサマリーメッセージを投稿します

各段階は、入力と出力が明確な独立した関数です。

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)

ステージ1：メールの読み取り

Gmail APIを使って新しいメールを取得します。google-api-python-clientライブラリが認証とAPI呼び出しを処理します。まだ確認していないメッセージをポーリングします。

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

ステージ2：アクションアイテムの抽出

メールの内容をLLMに渡し、構造化されたJSON形式でアクションアイテムを抽出するよう依頼します。response_formatを使うと、信頼性の高いJSON出力を取得できます。

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
    ]

ステージ3：Trelloカードの作成

Trello REST APIでは、単純なPOSTリクエストでカードを作成できます。APIキー、トークン、カードを作成するリストのIDが必要です。

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

ステージ4：Slack通知の送信

Slack SDKを使って、整形したサマリーメッセージをチャンネルに送信します。抽出して作成したアクションアイテムの内容が明確に伝わるメッセージにしてください。

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

各ステージのエラー処理

パイプラインの堅牢性は、エラー処理の堅牢性によって決まります。各ステージは独立して失敗する可能性があります。各呼び出しをラップしてエラーをログに記録し、パイプラインを継続するか中止するかを判断してください。

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
    )

パイプラインの構造化ログ

構造化ログを使用すると、パイプラインの実行履歴を検索できます。パイプラインの開始、各ステージの完了、最終結果を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})

メインパイプラインランナー

メインランナーがすべてをつなぎます。新着メールをポーリングし、それぞれをパイプラインで処理した後、再処理を防ぐために既読にします。

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)

設定管理

すべての API 認証情報とパイプライン設定は、コードではなく環境変数に保存します。起動時に読み込み、必須キーが存在することを検証します。

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

パイプラインのテスト

エンドツーエンドで実行する前に、各パイプラインステージをモックデータで個別にテストします。これにより、API クレジットを消費したり実際の 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()

理解度チェック：マルチアプリパイプライン

マルチアプリ自動化パイプラインの構築についての理解度を確認します。

パイプラインのまとめ

Gmail による検出、LLM ベースの抽出、Trello カードの作成、Slack 通知を備えた、完全なマルチアプリ自動化パイプラインを構築しました。主な設計原則は、明確なインターフェースを持つステージ分離、各ステップでの堅牢なエラー処理、構造化ロギング、環境変数による設定管理です。

よくある質問

「複数アプリの自動化パイプラインを構築する」レッスンは無料ですか？

はい。「複数アプリの自動化パイプラインを構築する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「複数アプリの自動化パイプラインを構築する」で何を学びますか？

エージェントが管理するツール呼び出しで Gmail → Slack → Google Sheets を連携します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「複数アプリの自動化パイプラインを構築する」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. トリガー・アクション型エージェントのパターン
  2. エージェントとWebhookの接続
  3. スケジューリングとCronベースのエージェント
  4. 複数アプリの自動化パイプラインを構築する
← AI Agentsに戻る