Python dla programisty JS: Agent do Gmail i kalendarza z FastAPI oraz LangGraph

Dla programisty, którego codziennymi narzędziami są Node.js, React, Express i Next, Python może wydawać się obcym terytorium – z innymi konwencjami i ekosystemem, który trzeba poznać od zera. Dobrym sposobem na przezwyciężenie tego jest stworzenie czegoś rzeczywistego i przydatnego, a nie aplikacji typu todo czy kalkulatora. Ten projekt to agent asystenta osobistego: sztuczna inteligencja, która łączy się z Gmail i Google Calendar, rozumie pytania w prostym języku angielskim i sama decyduje, co robić.

Zapytaj ją, jakie e-maile przyszły w tym tygodniu, a ona przeczyta skrzynkę odbiorczą. Poproś ją o umieszczenie spotkania rano jutro, a pojawi się wydarzenie w kalendarzu. Poproś o odpowiedź na odmowę pracy, a ona napisze szkic, który nigdy nie jest wysyłany bezpośrednio, dzięki czemu osoba musi go najpierw przejrzeć.

Od RAG do agentów

Generowanie wzbogacone o wyszukiwanie umożliwia botowi czatowemu odpowiadanie na pytania z twoich własnych dokumentów zamiast wymyślać odpowiedzi. Budowa takiego botu uczy, jak zachowują się modele językowe, jak działają embeddingi oraz bazy danych wektorowych takie jak ChromaDB, oraz jak przesyłać odpowiedzi za pomocą zdarzeń wysyłanych przez serwer. Mimo to RAG jest pasywne: przyjmuje pytanie, wyszukuje w dokumentach i odpowiada. W rzeczywistości nic się nie dzieje.

Agent nie poprzestaje na odpowiadaniu – działa. Różnicę najłatwiej zobaczyć porównując je obok siebie:

RAG: Question → Search docs → Answer
Agent: Question → Think → Pick a tool → Use it → Think again → Answer
 ↑___________________________________|
 (loops until the task is done)

Agent rozumuje w pętli i posiada zestaw narzędzi, którymi są funkcje, które może wywołać, a sam decyduje, które z nich pasuje do żądania. Pytanie dotyczące spotkań kieruje je do narzędzia kalendarza. Prośba o stworzenie notatki dla menedżera kieruje je do narzędzia do tworzenia projektów w Gmail. Żądanie wymagające obu rozwiązań sprawia, że wywołuje je kolejno. Za pierwszym razem, gdy to działa, wydaje się to nieco jak magia.

Podczas rozwoju cała infrastruktura działa bezpłatnie i offline, wykorzystując lokalny model, a pojedyncza zmienna środowiskowa zastępuje Claude w środowisku produkcyjnym.

Ustawienie środowiska

Zawsze aktywuj środowisko wirtualne przed instalacją lub uruchamianiem czegokolwiek, ponieważ zapomnienie jest najczęstszą przyczyną zamieszania.

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

Najmniejszy serwer FastAPI

FastAPI pełni rolę, jaką w świecie JavaScript pełni Express. Przykład „Hello World” wygląda tak:

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}"}

Rozpocznij od uvicorn:

uvicorn main:app --reload

Flaga --reload jest odpowiednikiem nodemon – restartuje serwer za każdym razem, gdy zmieni się plik. Szybka prośba sprawdza punkt końcowy:

curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "hello"}'

Łączenie z Gmail

Teraz sprawy stają się poważniejsze, ponieważ agent komunikuje się z rzeczywistym kontem Gmail. Najpierw zajmujemy się stroną Google Cloud:

1. Otwórz konsolę Google Cloud i utwórz nowy projekt.
2. Włącz zarówno API Gmail, jak i API Google Calendar.
3. Stwórz uprawnienia OAuth typu Desktop App.
4. Pobierz plik JSON i zapisz go jako credentials.json.
5. W zakładce OAuth consent screen, w sekcji Audience, dodaj swój adres Gmail jako użytkownika testowego.

Następnie zainstaluj biblioteki klienta Google:

pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client

Narzędzie Gmail tworzy jedynie projekty wiadomości i nigdy ich nie wysyła. Pracownik nie powinien mieć uprawnień do wysyłania maili w twoim imieniu bez etapu weryfikacji.

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

Pierwszy raz, gdy uruchamiasz aplikację, okno przeglądarki prosi o zalogowanie się przez Google i ostrzega, że aplikacja nie została zweryfikowana. Wybierz opcję „Zaawansowane”, a następnie przejdź do aplikacji. Jest to normalne w przypadku oprogramowania, które jest nadal w fazie rozwoju. Dane logowania trafiają potem do pliku token.json, dzięki czemu logowanie odbywa się tylko raz.

Jedna zasada bezpieczeństwa wymaga stanowczego podkreślenia: nigdy nie dodawaj plików token.json ani credentials.json do repozytorium git. Narzędzie skanowania tajemnic GitHub często blokuje takie operacje, ale lepiej nie polegać na tym rozwiązaniu. Natychmiast dodaj te pliki do pliku .gitignore:

token.json
credentials.json
.env
venv/
__pycache__/

Łączenie z kalendarzem Google

Procedura jest taka sama jak w przypadku Gmail: należy się zalogować, a następnie wywołać 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}"

Sam agent

LangGraph łączy wszystkie elementy ze sobą. Zainstaluj go wraz z integracjami modeli:

pip install langgraph langchain-anthropic langchain-core langchain-ollama

Agent posiada cztery narzędzia, które są zwykłymi funkcjami Pythona, które LangGraph może wywołać automatycznie, gdy tylko posiadają dekorator narzędzia:

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

Wybór llama3.1 zamiast llama3.2

Aby skorzystać z narzędzia, model musi generować strukturyzowany JSON, który LangGraph może przetworzyć, a mniejsze modele mają z tym trudności. Model 3B llama3.2 ciągle zwracał surowy JSON zamiast faktycznie wywołać narzędzie, podczas gdy model 8B llama3.1 radzi sobie z tym niezawodnie. W przypadku agentów rozmiar modelu ma większe znaczenie niż w zwykłej rozmowie.

Strumieniowanie rozumowania agenta

Jedną z najlepszych cech LangGraph jest możliwość obserwowania, jak agent krok po kroku wykonuje zadanie: jakie narzędzie wybrał, jaki wynik otrzymał i jak doszedł do odpowiedzi. Zdarzenia wysyłane przez serwer przekazują ten strumień do interfejsu użytkownika w czasie rzeczywistym. Odpowiedni kod znajduje się w pliku 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")

W interfejsie użytkownik widzi następującą sekwencję:

🧠 Using tool: tool_read_emails...
📦 Tool result ▼  (click to expand raw data)
💬 Here are your 5 latest emails...

Widok tego, jak agent myśli na głos, jest przekonującą demonstracją. Skrypt testowy uruchomiony na lokalnym modelu wyświetla pełny ślad działania:

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

Wynik opiera się na rzeczywistych adresach e-mail i prawdziwych wydarzeniach kalendarzowych, przy czym agent sam wybiera odpowiedni narzędzie bez żadnej pomocy, wszystko to na laptopie bez połączenia z Internetem.

Lekcje, które warto zapamiętać

• Dla agentów rozmiar modelu ma większe znaczenie, niż się spodziewano. RAG radzi sobie z małymi modelami, ponieważ czyta jedynie kontekst i tworzy streszczenie. Wywoływanie narzędzi wymaga ważnego, zstrukturyzowanego JSON, a małe modele często błędnie interpretują ten format. Różnica między modelem 8B a 3B jest tutaj ogromna.
• Utrzymuj prompty krótkie i jasne. Długi zbiór szczegółowych instrukcji jest w większości ignorowany przez małe, lokalne modele, podczas gdy około sześć wierszy z listą punktową zasad działa znacznie lepiej.
• Domyślną opcją powinien być tryb tylko projektów. Ustalenie, że create_draft to jedyne możliwe działanie związane z e-mailem, pozwala ludziom zachować kontrolę. Agent, który może samodzielnie wysyłać e-maile, stanowi zagrożenie, dlatego należy pozwolić ludziom sprawdzić treść przed jej wysłaniem.
• Środowiska wirtualne karzą za zapominalstwo. Gdy tylko pojawi się błąd ModuleNotFoundError, rozwiązaniem jest niemal zawsze ponowne aktywowanie środowiska wirtualnego.
• Python jest łatwiejszy w użyciu, niż wydaje się programiście JavaScript. Nie ma średników, nie ma nawiasów klamrowych, a struktura kodu wynika z wcięć. Po około tygodniu codziennego używania język schodzi na dalszy plan, a problem staje się głównym wyzwaniem.

Uruchamianie ukończonego projektu

Poniższe polecenia klonują ukończony projekt i uruchamiają go:

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

Dalej, gdzie iść

• Narzędzie do wyszukiwania w sieci, aby agent mógł szukać informacji online i je podsumowywać.
• Historia rozmów, ponieważ każde pytanie jest obecnie pozbawione kontekstu, a pamięć sesji umożliwiłaby naturalne kontynuacje.
• Więcej narzędzi, takich jak notatki, przypomnienia, wiadomości w Slacku lub problemy w GitHubie. Zasada pozostaje niezmienna: napisz funkcję, ozdób ją tagiem @tool i dodaj do agenta.
• Wdrożenie w środowisku produkcyjnym, przy użyciu Dockera, modelu Claude zamiast lokalnego oraz platformy takiej jak Railway lub Render.

Agenci to nie magia. Są to modele językowe z funkcjami, do których mogą się odwoływać, a gdy to zadziała, otwiera się całe pole możliwości. To właśnie podczas tworzenia czegoś następuje nauka.

Podobne materiały

• Parallelne agenty LangGraph, które recenzują pull requests za pomocą webhooków FastAPI — Jak webhook, kolejka Redis oraz cztery agenty LangGraph przekształcają pull request w uszeregowaną, strukturyzowaną recenzję AI w ciągu około 30 sekund, przy użyciu ngrok do testów lokalnych.
• Triaging zapytań o podporę za pomocą wielu agentów z LangGraph i Ollama — Stworzenie grafu triagingu zapytań z trzema agentami przy użyciu LangGraph: klasyfikatora, odpowiedziwcy oraz ludzkiego moderatora, działających na lokalnym Ollama lub przez API z jedną zmienioną linią kodu.