JS开发者必备的Python：基于FastAPI与LangGraph的Gmail及日历自动化工具
通过构建一个仅用于草稿的个人助手来学习 Python：FastAPI、用于 Gmail 和日历的 Google OAuth、LangGraph 工具、本地 Ollama 模型以及 SSE 流式传输。
来源: Learn Python by Building an AI Agent That Reads Emails and Calendar
对于那些日常工作中主要使用 Node.js、React、Express 和 Next 的开发者来说，Python 似乎是一片陌生的领域，其规范和生态系统都需要从零开始学习。一个不错的切入点是为自己构建一些真正有用的事物，而不是简单的待办清单应用或计算器。这个项目就是一个个人助手智能体：它能够连接 Gmail 和 Google 日历，理解通俗的英文提问，并自行决定该执行什么操作。
可以询问它本周收到了哪些邮件，它会读取收件箱内容。告诉它明早安排一次站会，相应的日历事件就会生成。若需要为求职被拒的情况撰写回复，它会先生成草稿，但绝不会直接发送，始终由人工先进行审核。
从 RAG 到智能体
检索增强生成技术能让聊天机器人根据用户自身的文档来回答问题，而非凭空编造内容。构建此类系统有助于了解大语言模型的行为机制、嵌入向量以及ChromaDB等向量数据库的运作原理，以及如何通过服务器发送事件来传输响应。不过，RAG属于被动式技术：它接收问题、搜索文档并给出答案，现实中不会产生其他任何变化。
而智能体则不仅限于回答问题，还会主动采取行动。两者的区别通过对比最容易看清楚：
RAG: Question → Search docs → Answer
Agent: Question → Think → Pick a tool → Use it → Think again → Answer
↑___________________________________|
(loops until the task is done)
智能体通过循环进行推理，并拥有一组工具，即它可以调用的函数，它会自行判断哪个函数适合当前请求。如果是关于会议的问题，它会调用日历工具；如果是要求给经理起草便条的请求，它会调用Gmail的草稿工具；如果需要同时使用两种功能，它会按顺序依次调用它们。第一次看到这种效果时，确实会感觉有点像魔法。
在开发阶段，整个系统都在离线状态下运行，使用本地模型，而通过一个环境变量可以在生产环境中用Claude替代原有模型。
环境设置
在安装或运行任何程序之前，务必先激活虚拟环境，因为忘记这一点是最常见的混淆来源。
python3 -m venv venv
source venv/bin/activate # Mac/Linux
# you'll see (venv) in your terminal — that means it's active
pip install fastapi uvicorn python-dotenv
最小的 FastAPI 服务器
FastAPI 扮演着 JavaScript 世界中 Express 的角色。一个“Hello World”示例如下：
from fastapi import FastAPI
from dotenv import load_dotenv
load_dotenv() # reads your .env file
app = FastAPI()
@app.get("/")
def root():
return {"message": "Agent is running!"}
@app.post("/ask")
def ask(body: dict):
question = body.get("question")
return {"answer": f"You asked: {question}"}
先用 uvicorn 启动它：
uvicorn main:app --reload
--reload 参数是 nodemon 的对应选项，会在文件发生变化时重新启动服务器。通过一个快速请求即可测试该接口：
curl -X POST http://localhost:8000/ask \
-H "Content-Type: application/json" \
-d '{"question": "hello"}'
连接 Gmail
现在进入实际操作阶段，因为该代理程序会与真实的 Gmail 账户进行交互。首先处理 Google Cloud 端的操作：
- 打开 Google Cloud 控制台并创建一个新项目。
- 启用 Gmail API 和 Google Calendar API。
- 创建类型为“桌面应用”的 OAuth 凭证。
- 下载 JSON 文件并将其保存为
credentials.json。 - 在 OAuth 同意界面中，于“受众”部分添加自己的 Gmail 地址作为测试用户。
接着安装 Google 客户端库：
pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client
Gmail工具仅能创建草稿，而不会实际发送邮件。未经审核流程，客服人员不应拥有代您发送邮件的权限。
import os
import base64
import re
from email.mime.text import MIMEText
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
SCOPES = [
'https://www.googleapis.com/auth/gmail.readonly',
'https://www.googleapis.com/auth/gmail.compose',
]
def get_gmail_service():
creds = None
if os.path.exists('token.json'):
creds = Credentials.from_authorized_user_file('token.json', SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
creds = flow.run_local_server(port=0)
with open('token.json', 'w') as f:
f.write(creds.to_json())
return build('gmail', 'v1', credentials=creds)
def read_emails(max_results=5):
service = get_gmail_service()
results = service.users().messages().list(
userId='me', maxResults=max_results, labelIds=['INBOX']
).execute()
emails = []
for msg in results.get('messages', []):
message = service.users().messages().get(
userId='me', id=msg['id'], format='full'
).execute()
headers = message['payload']['headers']
emails.append({
'subject': next((h['value'] for h in headers if h['name'] == 'Subject'), ''),
'from': next((h['value'] for h in headers if h['name'] == 'From'), ''),
'date': next((h['value'] for h in headers if h['name'] == 'Date'), ''),
})
return emails
def create_draft(to: str, subject: str, body: str):
# Extract clean email address from "Name <email>" format
match = re.search(r'[\w\.-]+@[\w\.-]+\.\w+', to)
if match:
to = match.group(0)
service = get_gmail_service()
message = MIMEText(body)
message['to'] = to
message['subject'] = subject
raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
draft = service.users().drafts().create(
userId='me', body={'message': {'raw': raw}}
).execute()
return f"Draft created for {to} — subject: '{subject}'"
首次运行时，浏览器窗口会要求你使用 Google 登录，并提示该应用尚未通过验证。请选择“高级”选项，然后进入该应用。对于仍在开发中的软件来说，这是正常现象。之后凭证会被保存到 token.json 中，这样只需登录一次即可。
有一条安全规则必须明确强调：绝不要将 token.json 或 credentials.json 提交到 git 中。GitHub 的机密信息扫描通常会阻止此类提交，但最好还是不要依赖这一点。应立即将这些文件添加到 .gitignore 中：
token.json
credentials.json
.env
venv/
__pycache__/
连接 Google 日历
流程与 Gmail 相同：先进行身份验证，然后再调用 API。
from datetime import datetime, timedelta
from googleapiclient.discovery import build
# ... (same auth setup as gmail_tool.py)
def get_upcoming_events(days=7, max_results=10):
service = get_calendar_service()
now = datetime.utcnow()
future = now + timedelta(days=days)
events = service.events().list(
calendarId='primary',
timeMin=now.isoformat() + 'Z',
timeMax=future.isoformat() + 'Z',
maxResults=max_results,
singleEvents=True,
orderBy='startTime'
).execute().get('items', [])
return [{
'title': e.get('summary', 'No title'),
'start': e['start'].get('dateTime', e['start'].get('date')),
'end': e['end'].get('dateTime', e['end'].get('date')),
'location': e.get('location', ''),
'attendees': [a['email'] for a in e.get('attendees', [])],
} for e in events]
def create_event(title, date, start_time, end_time, description='', attendees=[]):
service = get_calendar_service()
event = {
'summary': title,
'description': description,
'start': {'dateTime': f'{date}T{start_time}:00', 'timeZone': 'America/Los_Angeles'},
'end': {'dateTime': f'{date}T{end_time}:00', 'timeZone': 'America/Los_Angeles'},
}
if attendees:
event['attendees'] = [{'email': e} for e in attendees]
created = service.events().insert(
calendarId='primary', body=event, sendUpdates='none'
).execute()
return f"Event '{title}' created on {date} from {start_time} to {end_time}"
代理本身
LangGraph负责将各组件连接起来。请在模型集成中一并安装它：
pip install langgraph langchain-anthropic langchain-core langchain-ollama
该代理拥有四种工具，这些工具都是普通的 Python 函数，只要加上工具装饰器，LangGraph 就能自动调用它们：
import os
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from gmail_tool import read_emails, create_draft
from calendar_tool import get_upcoming_events, create_event
load_dotenv()
# ── Dev/prod switch ───────────────────────────────────────────────
ENV = os.getenv("APP_ENV", "development")
if ENV == "production":
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-sonnet-4-6", api_key=os.getenv("ANTHROPIC_API_KEY"))
print("🤖 Using Claude (production)")
else:
from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
print("🤖 Using Ollama llama3.1 (development)")
# ── Tools (functions the agent can call) ─────────────────────────
@tool
def tool_read_emails(max_results: int = 5) -> str:
"""Read latest emails from Gmail."""
emails = read_emails(max_results)
if not emails:
return "No emails found."
return "\n---\n".join(
f"From: {e['from']}\nSubject: {e['subject']}\nDate: {e['date']}"
for e in emails
)
@tool
def tool_create_draft(to: str, subject: str, body: str) -> str:
"""Create a Gmail draft. Args: to=email address, subject=subject line, body=email text."""
return create_draft(to, subject, body)
@tool
def tool_get_events(days: int = 7) -> str:
"""Get upcoming Google Calendar events."""
events = get_upcoming_events(days)
if not events:
return f"No events in the next {days} days."
return "\n---\n".join(
f"📅 {e['title']}\n {e['start']} → {e['end']}"
+ (f"\n 📍 {e['location']}" if e['location'] else "")
for e in events
)
@tool
def tool_create_event(title: str, date: str, start_time: str, end_time: str, description: str = "", attendees: str = "") -> str:
"""Create a Google Calendar event. Date: YYYY-MM-DD. Time: HH:MM."""
attendee_list = [a.strip() for a in attendees.split(",")] if attendees else []
return create_event(title, date, start_time, end_time, description, attendee_list)
# ── Agent ─────────────────────────────────────────────────────────
agent = create_react_agent(
model=llm,
tools=[tool_read_emails, tool_create_draft, tool_get_events, tool_create_event],
prompt="""You are a personal assistant. Use tools to help with emails and calendar.
Rules:
- ALWAYS use a tool to answer. Never guess.
- For emails: use tool_read_emails
- For drafts: use tool_create_draft
- For calendar: use tool_get_events
- For new events: use tool_create_event"""
)
def ask_agent(question: str) -> str:
result = agent.invoke({"messages": [HumanMessage(content=question)]})
return result['messages'][-1].content
选择 llama3.1 而非 llama3.2
工具调用要求模型输出 LangGraph 能解析的结构化 JSON，而较小规模的模型难以做到这一点。30 亿参数的 llama3.2 总是返回原始 JSON，未能真正调用工具，而 80 亿参数的 llama3.1 则能稳定地完成该任务。对于智能体而言，模型规模的重要性远超普通聊天场景。
实时流式展示智能体的推理过程
LangGraph 最出色的功能之一就是能够逐步查看智能体处理任务的过程：它选择了哪个工具、得到了什么结果，以及是如何确定最终答案的。服务器发送的事件会实时将这一流程推送到前端。相关代码位于 main.py 中：
# main.py
import json
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage
from agent import agent
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
@app.get("/api/ask")
async def ask(q: str):
def stream():
for chunk in agent.stream(
{"messages": [HumanMessage(content=q)]},
stream_mode="updates"
):
# Agent deciding which tool to use
if "agent" in chunk:
for msg in chunk["agent"].get("messages", []):
if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
yield f"data: {json.dumps({'type': 'thinking', 'text': f'Using tool: {tc[\"name\"]}...'})}\n\n"
elif hasattr(msg, "content") and msg.content:
yield f"data: {json.dumps({'type': 'answer', 'text': msg.content})}\n\n"
# Tool result coming back
if "tools" in chunk:
for msg in chunk["tools"].get("messages", []):
if hasattr(msg, "content") and msg.content:
yield f"data: {json.dumps({'type': 'tool_result', 'text': str(msg.content)[:300]})}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream")
在界面中，用户会看到如下序列：
🧠 Using tool: tool_read_emails...
📦 Tool result ▼ (click to expand raw data)
💬 Here are your 5 latest emails...
看到代理程序大声思考的过程本身就是一种有说服力的演示。针对本地模型运行的测试脚本会输出完整的执行轨迹：
python3 test_agent.py
🤖 Using Ollama llama3.1 (development)
🤖 Testing agent...
Test 1: Read emails
Here are your latest 3 emails:
1. From Bilt <notifications@members.bilt.com> — Your ALL Accor status match unlocks August 1st
2. From Vanguard <vanguard@...> — Your Vanguard site is getting an upgrade
3. From Kaggle <no-reply@kaggle.com> — Thank You for Making the 5-Day AI Agents Intensive a Success!
Test 2: Check calendar
You have a meeting scheduled for July 31st, 1pm-2pm.
Test 3: Create draft email
✅ Draft created for test@example.com — subject: 'Hello'
Test 4: Create calendar event
✅ Event 'Agent Test Meeting' created on 2026-07-31 from 14:00 to 15:00
该系统使用真实的电子邮件和日历事件，智能体无需协助即可自行选择合适的工具，且所有操作都在无网络连接的笔记本电脑上完成。
值得保留的经验
- 对智能体而言，模型规模的重要性超出预期。RAG技术能够应对小型模型，因为它只需读取上下文并进行总结。而调用工具则需要有效的结构化JSON数据，小型模型往往会在格式处理上出现错误，因此8B模型与3B模型之间的差距在此处十分明显。
- 提示语应简短且明确。冗长复杂的指令大多会被小型本地模型忽略，而用项目符号列出规则的六行左右的内容效果要好得多。
- 默认应仅允许草稿功能。将
create_draft设为唯一的邮件发送操作，这样就能由人类掌控全局。若允许智能体自主发送邮件，将会带来风险，因此必须在内容发送前由人类进行审核。
ModuleNotFoundError 时，解决办法几乎总是重新激活该虚拟环境。运行已完成的项目
以下命令可用于克隆已完成的项目并启动它：
git clone https://github.com/TpPrachi/personal_assistant_agent.git
cd personal_assistant_agent
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
# Pull the model (8B, better tool calling)
ollama pull llama3.1
ollama serve
# Set up Google credentials (see README for full steps)
# Then authenticate
python3 test_gmail.py
# Run the agent
python3 test_agent.py
下一步该去哪里
- 一个网络搜索工具，这样智能体就可以在网上查找信息并加以总结。
- 对话历史记录，因为当前每个问题都是无状态的，而会话记忆功能可以让对话更自然地延续。
- 更多工具，如笔记、提醒、Slack 消息或 GitHub 问题。方法始终不变：编写一个函数，用
@tool标记它，然后将其添加到智能体中。 - 生产环境部署，使用 Docker、Claude 作为模型替代本地模型，并选择 Railway 或 Render 之类的托管服务。
智能体并非魔法。它们只是拥有可调用函数的大型语言模型，一旦这些功能得以运用，整个领域就会迎刃而解。真正的学习发生在构建实物的过程中。