Python для разработчика JS: агент для Gmail и календаря с FastAPI и LangGraph

Для разработчика, который ежедневно использует Node.js, React, Express и Next, Python может показаться чужой сферой с другими правилами и экосистемой, которую придётся изучать с нуля. Хороший способ преодолеть это — создать что-то реальное и полезное, а не просто приложение для списка задач или калькулятор. Этот проект — это персональный помощник-агент: ИИ, который подключается к Gmail и Google Calendar, понимает вопросы на простом английском языке и самостоятельно определяет, что делать.

Спросите у него, какие письма пришли на этой неделе, и он прочитает почтовый ящик. Попросите организовать утреннюю встречу, и появится запись в календаре. Попросите написать ответ на отказ от работы, и он составит черновик, который никогда не отправляется напрямую, чтобы человек всегда мог его проверить.

От 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 выполняет ту же роль, что Express в мире JavaScript. Пример кода «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:

1. Откройте консоль Google Cloud и создайте новый проект.
2. Включите API Gmail и API Google Calendar.
3. Создайте учётные данные OAuth типа Desktop App.
4. Скачайте файл JSON и сохраните его как credentials.json.
5. На экране согласия OAuth в разделе Audience добавьте свой адрес 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

Для вызова инструментов необходимо, чтобы модель генерировала структурированный JSON, который может парсить LangGraph, но более мелкие модели с трудом с этим справляются. Модель размером 3 млрд параметров llama3.2 продолжала возвращать сырой JSON вместо того, чтобы действительно вызвать инструмент, тогда как модель размером 8 млрд параметров 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, решением почти всегда является повторная активация venv.
• Python оказывается проще, чем кажется разработчикам JavaScript: нет точек с запятой, нет фигурных скобок, а структура кода определяется отступами. После примерно недели ежедневного использования язык отходит на второй план, а проблемы становятся главным вниманием.

Запуск готового проекта

Следующие команды клонируют готовый проект и запускают его:

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.

Агенты — это не магия. Это LLM с функциями, которые им разрешено вызывать, и как только это начинает работать, открывается весь спектр возможностей. Именно при создании чего-то происходит обучение.