0Pricing
AI Agents · 课时

以编程方式读取和发送电子邮件

列出邮件、获取邮件正文并发送 MIME 电子邮件。

以编程方式读取和发送电子邮件 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

使用 Gmail API 列出邮件

读取邮件的第一步是列出符合条件的邮件。service.users().messages().list() 会返回邮件 ID 和会话 ID，而不是完整内容。然后，您需要逐封获取邮件。这种两步模式可以让列出操作保持快速。

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

获取完整邮件

使用 service.users().messages().get() 获取完整的邮件内容。format 参数控制返回的数据量：'full' 包含标头和正文，'metadata' 仅返回标头，'minimal' 只返回 ID 和标签。

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

提取邮件标头

标头（发件人、收件人、主题、日期）存储在 message['payload']['headers'] 中，是由 {'name': ..., 'value': ...} 字典组成的列表。请编写一个按名称提取标头的辅助函数——您会经常使用它。

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"]}')

解码邮件正文（base64）

Gmail API 中的邮件正文采用 base64url 编码——这是 base64 的 URL 安全变体，其中 + 会变成 -，/ 会变成 _。请使用 base64.urlsafe_b64decode() 进行解码。同时处理简单的单部分邮件和多部分邮件。

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

递归处理多部分邮件

复杂邮件（包含附件、内嵌图片或混合内容）采用嵌套的多部分结构。正文部分可以任意深度嵌套。使用递归函数遍历部分树，即可处理所有情况。

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

将邮件标记为已读

处理邮件后，您的智能体应通过移除 UNREAD 标签将其标记为已读。请使用 service.users().messages().modify()，并传入 removeLabelIds=['UNREAD']。您也可以添加 PROCESSED 等标签，以跟踪已由智能体处理的邮件。

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

使用 MIMEText 编写邮件

要发送邮件，请先使用 Python 的标准 email 库将其编写为 MIME 消息。然后对原始字节进行 base64url 编码，并通过 POST 将其发送到 Gmail API。MIMEText 会负责正确编码邮件正文。

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

使用 Gmail API 发送邮件

使用 service.users().messages().send() 发送编写好的邮件。userId='me' 参数表示经过身份验证的用户。API 会返回已发送的邮件及其 ID 和会话 ID。

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)

创建并发送邮件草稿

智能体不必立即发送邮件，也可以创建草稿供人工审核。请使用 service.users().drafts().create()。然后，人工可以在 Gmail 界面中审核并发送草稿。对于任何需要人工批准的邮件，这是推荐的模式。

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

批量处理多封邮件

处理大量邮件时，不要在紧密循环中逐封获取邮件——这样会触及配额限制。请使用带有短暂延迟的受控循环，或使用 Gmail API 的批量请求功能，将多个操作合并到一次 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

回复邮件（在同一会话中）

要发送同一会话中的回复，请将 In-Reply-To 和 References 标头设置为原始邮件的 Message-ID 标头，并将 threadId 传递给发送调用。这样可以让回复保留在同一个 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
    }

快速检查：base64 邮件正文

检验您对 Gmail API 邮件处理的理解。

读取和发送邮件回顾

现在，您的智能体已经可以通过程序读取和发送邮件：

  • 列出：messages().list(q='is:unread') 返回 ID；使用 Gmail 搜索语法
  • 获取：messages().get(id=..., format='full') 返回完整邮件
  • 解析标头：从 payload.headers 中提取发件人、主题和日期
  • 解码正文：对于文本，使用 base64.urlsafe_b64decode(data)；对于多部分邮件，递归处理各部分
  • 发送：使用 MIMEText 编写邮件，进行 base64url 编码，再通过 messages().send() 使用 POST 发送
  • 在会话中回复：设置 In-Reply-To 标头，并在发送正文中加入 threadId

常见问题解答

「以编程方式读取和发送电子邮件」课时是免费的吗？

是的 — 「以编程方式读取和发送电子邮件」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「以编程方式读取和发送电子邮件」这节课中我会学到什么？

列出邮件、获取邮件正文并发送 MIME 电子邮件。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「以编程方式读取和发送电子邮件」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 通过 API 连接 Gmail
  2. 以编程方式读取和发送电子邮件
  3. 创建与查询日历事件
  4. 构建简单的电子邮件助手代理
← 返回 AI Agents