AI Agents · 강의

다중 앱 자동화 파이프라인 구축

에이전트가 관리하는 도구 호출로 Gmail → Slack → Google Sheets 연결하기

다중 앱 자동화 파이프라인 구축은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

다중 앱 파이프라인

다중 앱 파이프라인은 여러 서비스를 하나의 자동화된 워크플로로 연결합니다. 이 과정에서는 다음을 구축합니다. Gmail 새 이메일 → 에이전트가 읽고 작업 항목 추출 → Trello 카드 생성 → Slack 알림 전송.

파이프라인 아키텍처

파이프라인은 네 단계로 구성됩니다.

트리거: Gmail 푸시 알림 또는 폴링으로 새 이메일 감지
추출: LLM이 이메일을 읽고 작업 항목 추출
생성: Trello API가 각 작업 항목에 대한 카드 생성
알림: 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 형식으로 작업 항목을 추출하도록 요청하십시오. 신뢰할 수 있는 JSON 출력을 얻으려면 response_format을 사용하십시오.

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/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“다중 앱 자동화 파이프라인 구축”에서 뭘 배우나요?

에이전트가 관리하는 도구 호출로 Gmail → Slack → Google Sheets 연결하기 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.

“다중 앱 자동화 파이프라인 구축” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기