Python для JS-распрацоўшчыка: агент для Gmail і Calendar на FastAPI і LangGraph

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

Запытайце яго, якія лісты прыйшлі за гэты тыдзень, і ён прачытае папку з некачанымі лістамі. Прабачце яму задаць заведамоць на завтра ранак, і ў календары з’явіцца відпаведны запіс. Запытайце яго напісаць адпаведзь на адмову у працевым месцы, і ён створыць чэрнавік, які ніколі не надсылае безпосерэдна, таму чалавек завжды спачатку яго пераглядае.

З RAG да агентаў

Тэхніка Retrieval-augmented generation дазваляе чат-боту адпавядаць на запытанні з вашых сабеўных дакументаў, а не выдумваць адпаведзі. Стварэнне такога чат-бота научае, як паводзяцца большыя мовныя мадэлі, як працуюць эмбедынгі і векторныя базы дадзэння, такія як 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. У роздзеле Audience на экране схвалення 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

Для вызвання інструменту модэль должна выдаваць структураваны JSON, які LangGraph можа распакаваць, а маленькія модэлі страждаюць з гэтым. Модэль 3B llama3.2 продовжвала вяртаць необработаны JSON заместо фактычнага вызвання інструменту, тады калі модэль 8B 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-ы з функцыямі, якія ім дазволеныя вызываць, і калі гэта запрацоўвае, ачынаецца цэлы простор для працы. Самэ працэўванне над чым-небудзь і ёсць месца навучэння.