0Pricing
AI Agents · Lekcja

Programowe odczytywanie i wysyłanie e-maili

Listowanie wiadomości, pobieranie treści wiadomości i wysyłanie wiadomości MIME.

Programowe odczytywanie i wysyłanie e-maili to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Wyświetlanie listy wiadomości za pomocą Gmail API

Pierwszym krokiem podczas odczytywania poczty jest wyświetlenie listy wiadomości spełniających określone kryteria. service.users().messages().list() zwraca identyfikatory wiadomości i wątków, a nie pełną treść. Następnie każdą wiadomość należy pobrać osobno. Taki dwuetapowy wzorzec pozwala zachować szybkość wywołania listy.

def list_messages(service, user_id='me', query='', max_results=10):
    results = service.users().messages().list(
        userId=user_id,
        q=query,          # Gmail search query
        maxResults=max_results
    ).execute()

    messages = results.get('messages', [])
    print(f'Found {len(messages)} messages')
    return messages

# Examples of Gmail search queries:
# 'is:unread' — unread messages
# 'from:boss@company.com is:unread' — unread from boss
# 'subject:invoice label:inbox' — invoices in inbox
# 'after:2026/05/01 has:attachment' — recent with attachments
messages = list_messages(gmail_service, query='is:unread label:inbox')

Pobieranie pełnej wiadomości

Do pobrania pełnej treści wiadomości należy użyć service.users().messages().get(). Parametr format określa ilość zwracanych danych: 'full' obejmuje nagłówki i treść, 'metadata' zwraca tylko nagłówki, a 'minimal' — wyłącznie identyfikatory i etykiety.

def get_message(service, message_id, user_id='me'):
    message = service.users().messages().get(
        userId=user_id,
        id=message_id,
        format='full'  # 'full', 'metadata', or 'minimal'
    ).execute()
    return message

# Fetch the first unread message
messages = list_messages(gmail_service, query='is:unread', max_results=1)
if messages:
    msg = get_message(gmail_service, messages[0]['id'])
    print('Thread ID:', msg['threadId'])
    print('Labels:', msg['labelIds'])
    print('Snippet:', msg['snippet'][:100])

Wyodrębnianie nagłówków wiadomości e-mail

Nagłówki (From, To, Subject, Date) są przechowywane w message['payload']['headers'] jako lista słowników {'name': ..., 'value': ...}. Należy napisać funkcję pomocniczą wyodrębniającą nagłówki według nazwy — będzie ona używana bardzo często.

def get_header(message, name):
    headers = message.get('payload', {}).get('headers', [])
    for h in headers:
        if h['name'].lower() == name.lower():
            return h['value']
    return ''

def extract_email_meta(message):
    return {
        'id': message['id'],
        'from': get_header(message, 'From'),
        'to': get_header(message, 'To'),
        'subject': get_header(message, 'Subject'),
        'date': get_header(message, 'Date'),
        'snippet': message.get('snippet', '')
    }

meta = extract_email_meta(msg)
print(f'From: {meta["from"]}')
print(f'Subject: {meta["subject"]}')
print(f'Date: {meta["date"]}')

Dekodowanie treści wiadomości e-mail (base64)

Treści wiadomości e-mail w Gmail API są zakodowane w formacie base64url — jest to bezpieczny dla adresów URL wariant base64, w którym + zamienia się na -, a / na _. Do dekodowania należy użyć base64.urlsafe_b64decode(). Należy obsługiwać zarówno wiadomości proste (jednoczęściowe), jak i wieloczęściowe.

import base64

def decode_body(data):
    if not data:
        return ''
    decoded_bytes = base64.urlsafe_b64decode(data + '==')
    return decoded_bytes.decode('utf-8', errors='replace')

def get_email_body(message):
    payload = message.get('payload', {})
    mime_type = payload.get('mimeType', '')

    # Simple (non-multipart) email
    if 'body' in payload and payload['body'].get('data'):
        return decode_body(payload['body']['data'])

    # Multipart email: find the text/plain or text/html part
    parts = payload.get('parts', [])
    for part in parts:
        if part.get('mimeType') == 'text/plain':
            return decode_body(part['body'].get('data', ''))

    # Fallback: try text/html
    for part in parts:
        if part.get('mimeType') == 'text/html':
            return decode_body(part['body'].get('data', ''))

    return message.get('snippet', '')

# --- demo ---
encoded = base64.urlsafe_b64encode(b'Hello from the agent!').decode().rstrip('=')
print('Decoded body:', decode_body(encoded))

message = {
    'payload': {
        'mimeType': 'multipart/alternative',
        'parts': [
            {'mimeType': 'text/plain', 'body': {'data': encoded}}
        ]
    }
}
print('Email body:', get_email_body(message))

Rekursywna obsługa wiadomości multipart

Złożone wiadomości (z załącznikami, obrazami osadzonymi lub mieszaną treścią) mają zagnieżdżoną strukturę multipart. Ich części mogą być zagnieżdżone na dowolną głębokość. Rekurencyjna funkcja przechodząca po drzewie części obsługuje wszystkie przypadki.

import base64

def extract_parts(payload, target_mime='text/plain'):
    parts_text = []
    mime_type = payload.get('mimeType', '')

    if mime_type == target_mime:
        data = payload.get('body', {}).get('data', '')
        if data:
            decoded = base64.urlsafe_b64decode(data + '==').decode('utf-8', errors='replace')
            parts_text.append(decoded)

    # Recurse into sub-parts
    for part in payload.get('parts', []):
        parts_text.extend(extract_parts(part, target_mime))

    return parts_text

def get_plain_text(message):
    payload = message.get('payload', {})
    texts = extract_parts(payload, 'text/plain')
    return '\n\n'.join(texts) if texts else message.get('snippet', '')

body_text = get_plain_text(msg)
print(f'Body ({len(body_text)} chars):', body_text[:200])

Oznaczanie wiadomości jako przeczytanych

Po przetworzeniu wiadomości agent powinien oznaczyć ją jako przeczytaną, usuwając etykietę UNREAD. Należy użyć service.users().messages().modify() z parametrem removeLabelIds=['UNREAD']. Można również dodawać etykiety, takie jak PROCESSED, aby śledzić wiadomości obsłużone przez agenta.

def mark_as_read(service, message_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'removeLabelIds': ['UNREAD']}
    ).execute()
    print(f'Marked {message_id} as read')

def add_label(service, message_id, label_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'addLabelIds': [label_id]}
    ).execute()

# Get label ID by name
def get_label_id(service, label_name, user_id='me'):
    labels = service.users().labels().list(userId=user_id).execute()
    for label in labels.get('labels', []):
        if label['name'].lower() == label_name.lower():
            return label['id']
    return None

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeUsers:
    def messages(self):
        return self
    def modify(self, **kwargs):
        print(f'[gmail api] messages.modify({kwargs})')
        return _Exec({'id': kwargs.get('id')})
    def labels(self):
        return self
    def list(self, **kwargs):
        return _Exec({'labels': [{'id': 'Label_1', 'name': 'Processed'}]})

class _FakeService:
    def users(self):
        return _FakeUsers()

service = _FakeService()
mark_as_read(service, 'msg_42')
add_label(service, 'msg_42', 'Label_1')
print('Label id for "Processed":', get_label_id(service, 'Processed'))

Tworzenie wiadomości e-mail za pomocą MIMEText

Aby wysłać wiadomość e-mail, należy najpierw utworzyć ją jako wiadomość MIME za pomocą standardowej biblioteki Pythona email. Następnie trzeba zakodować surowe bajty w formacie base64url i wysłać je metodą POST do Gmail API. MIMEText zajmuje się prawidłowym kodowaniem treści wiadomości.

import base64
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def create_message(sender, to, subject, body_text, body_html=None):
    if body_html:
        msg = MIMEMultipart('alternative')
        msg.attach(MIMEText(body_text, 'plain', 'utf-8'))
        msg.attach(MIMEText(body_html, 'html', 'utf-8'))
    else:
        msg = MIMEText(body_text, 'plain', 'utf-8')

    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject

    # Encode as base64url
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {'raw': raw}

message = create_message(
    sender='agent@yourcompany.com',
    to='recipient@example.com',
    subject='Weekly Summary',
    body_text='Hello,\n\nHere is your summary.\n\nBest,\nAgent'
)

# --- demo ---
print('Message keys:', list(message.keys()))
print('Base64 length:', len(message['raw']))

Wysyłanie wiadomości e-mail za pomocą Gmail API

Utworzoną wiadomość należy wysłać za pomocą service.users().messages().send(). Parametr userId='me' odnosi się do uwierzytelnionego użytkownika. API zwraca wysłaną wiadomość wraz z jej identyfikatorem i identyfikatorem wątku.

from googleapiclient.errors import HttpError

def send_message(service, message, user_id='me'):
    try:
        sent = service.users().messages().send(
            userId=user_id,
            body=message
        ).execute()
        print(f'Message sent! ID: {sent["id"]}')
        return sent
    except HttpError as e:
        import json
        body = json.loads(e.content.decode())
        print(f'Send failed ({e.resp.status}): {body.get("error", {}).get("message")}')
        return None

# Send the message
message = create_message(
    sender='me',
    to='team@company.com',
    subject='Agent Report',
    body_text='Processing complete. 42 tasks handled.'
)
send_message(gmail_service, message)

Tworzenie i wysyłanie wersji roboczych wiadomości e-mail

Zamiast wysyłać wiadomości natychmiast, agenty mogą tworzyć wersje robocze do sprawdzenia przez człowieka. Należy użyć service.users().drafts().create(). Następnie człowiek może sprawdzić i wysłać wersję roboczą w interfejsie Gmaila. Jest to zalecany wzorzec w przypadku każdej wiadomości e-mail wymagającej zatwierdzenia przez człowieka.

def create_draft(service, message, user_id='me'):
    draft = service.users().drafts().create(
        userId=user_id,
        body={'message': message}
    ).execute()
    print(f'Draft created: {draft["id"]}')
    return draft

def send_draft(service, draft_id, user_id='me'):
    sent = service.users().drafts().send(
        userId=user_id,
        body={'id': draft_id}
    ).execute()
    print(f'Draft sent as message: {sent["id"]}')
    return sent

# Create a draft for review
message = create_message(
    sender='me',
    to='client@example.com',
    subject='Proposal Follow-up',
    body_text='Dear Client,\n\nFollowing up on our proposal...'
)
draft = create_draft(gmail_service, message)
# Human reviews in Gmail, then agent sends:
# send_draft(gmail_service, draft['id'])

Przetwarzanie wsadowe wielu wiadomości e-mail

Podczas przetwarzania wielu wiadomości nie należy pobierać ich pojedynczo w ścisłej pętli — doprowadzi to do przekroczenia limitów wykorzystania. Należy użyć kontrolowanej pętli z krótkimi opóźnieniami albo funkcji żądań zbiorczych Gmail API, aby połączyć wiele operacji w jedno wywołanie HTTP.

import time

def process_unread_emails(service, max_emails=20):
    messages = list_messages(
        service,
        query='is:unread label:inbox',
        max_results=max_emails
    )

    processed = []
    for i, msg_ref in enumerate(messages):
        # Rate-limit: process max 5 per second
        if i > 0 and i % 5 == 0:
            time.sleep(1)

        msg = get_message(service, msg_ref['id'])
        meta = extract_email_meta(msg)
        body = get_plain_text(msg)

        result = {
            'id': msg['id'],
            'from': meta['from'],
            'subject': meta['subject'],
            'body_preview': body[:200]
        }
        processed.append(result)
        mark_as_read(service, msg['id'])

    return processed

Odpowiadanie na wiadomość e-mail (w wątku)

Aby wysłać odpowiedź w istniejącym wątku, należy ustawić nagłówki In-Reply-To i References na wartość nagłówka Message-ID oryginalnej wiadomości oraz przekazać threadId do wywołania wysyłającego. Dzięki temu odpowiedź pozostanie w tym samym wątku rozmowy Gmail.

import base64
from email.mime.text import MIMEText

def create_reply(original_message, reply_text, sender='me'):
    original_msg_id = get_header(original_message, 'Message-ID')
    to = get_header(original_message, 'From')
    subject = get_header(original_message, 'Subject')
    if not subject.startswith('Re:'):
        subject = 'Re: ' + subject

    msg = MIMEText(reply_text, 'plain', 'utf-8')
    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject
    msg['In-Reply-To'] = original_msg_id
    msg['References'] = original_msg_id

    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {
        'raw': raw,
        'threadId': original_message['threadId']  # keeps it in thread
    }

Szybki test: treść wiadomości e-mail w formacie base64

Sprawdź swoją wiedzę na temat obsługi wiadomości Gmail API.

Podsumowanie odczytywania i wysyłania wiadomości e-mail

Agent potrafi teraz programowo odczytywać i wysyłać wiadomości e-mail:

  • Wyświetlanie listy: messages().list(q='is:unread') zwraca identyfikatory; należy używać składni wyszukiwania Gmail
  • Pobieranie: messages().get(id=..., format='full') zwraca pełną wiadomość
  • Analizowanie nagłówków: należy wyodrębniać From/Subject/Date z payload.headers
  • Dekodowanie treści: base64.urlsafe_b64decode(data) w przypadku tekstu; w przypadku części multipart należy używać rekurencji
  • Wysyłanie: należy utworzyć wiadomość za pomocą MIMEText, zakodować ją w formacie base64url i wysłać metodą POST przez messages().send()
  • Odpowiadanie w wątku: należy ustawić nagłówek In-Reply-To oraz threadId w treści żądania wysyłania

Często zadawane pytania

Czy lekcja „Programowe odczytywanie i wysyłanie e-maili” jest bezpłatna?

Tak — pełny tekst „Programowe odczytywanie i wysyłanie e-maili” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Programowe odczytywanie i wysyłanie e-maili”?

Listowanie wiadomości, pobieranie treści wiadomości i wysyłanie wiadomości MIME. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „Programowe odczytywanie i wysyłanie e-maili”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Łączenie z Gmailem przez API
  2. Programowe odczytywanie i wysyłanie e-maili
  3. Tworzenie i wyszukiwanie wydarzeń w kalendarzu
  4. Tworzenie prostego agenta asystenta e-mail
← Powrót do AI Agents