Inicio / Artículos

pythonfastapilanggraphai-agentsgmail-apiollama

Python para desarrolladores de JS: Un agente para Gmail y Calendario con FastAPI y LangGraph

Aprende Python creando un asistente personal de borrador: FastAPI, Google OAuth para Gmail y Calendar, herramientas de LangGraph, modelos locales de Ollama y transmisión en tiempo real SSE.

Editor: REACTAPP.TOP10 min de lectura2243 palabras

Fuentes: Learn Python by Building an AI Agent That Reads Emails and Calendar

Para un desarrollador cuyas herramientas diarias son Node.js, React, Express y Next, Python puede parecer un territorio desconocido, con convenciones diferentes y un ecosistema del que hay que aprender desde cero. Una buena forma de superar esto es crear algo real y útil, y no solo una aplicación de tareas o una calculadora. Este proyecto es un agente asistente personal: una IA que se conecta a Gmail y Google Calendar, entiende preguntas en inglés sencillo y decide por sí misma qué hacer.

Pídele qué correos llegaron esta semana y leerá la bandeja de entrada. Dile que agende una reunión para la mañana siguiente y aparecerá un evento en el calendario. Pídele que escriba una respuesta a un rechazo laboral y elaborará un borrador, el cual nunca envía directamente, de modo que siempre hay alguien que lo revisa primero.

De RAG a agentes

La generación mejorada por recuperación de información permite que un chatbot responda a preguntas a partir de sus propios documentos en lugar de inventar respuestas. Desarrollar uno enseña cómo se comportan los LLM, cómo funcionan los embeddings y las bases de datos vectoriales como ChromaDB, y cómo transmitir respuestas mediante eventos enviados por el servidor. Sin embargo, RAG es pasivo: recibe una pregunta, busca en los documentos y responde. No ocurre nada más en el mundo.

Un agente no se limita a responder; también actúa. La diferencia se observa más fácilmente comparándolos lado a lado:

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

Un agente razona en un bucle y dispone de un conjunto de herramientas, que son funciones que puede llamar, y decide por sí mismo cuál se adapta a la solicitud. Una pregunta sobre reuniones lo dirige a la herramienta de calendario. Una solicitud para redactar una nota a un gerente lo envía a la herramienta de borradores de Gmail. Una solicitud que necesita ambas cosas hace que las llame en secuencia. La primera vez que esto funciona parece un poco mágico.

Durante el desarrollo, toda la estructura funciona de forma gratuita y sin conexión, utilizando un modelo local, y una sola variable de entorno sustituye a Claude para el entorno de producción.

Configuración del entorno

Siempre active el entorno virtual antes de instalar o ejecutar cualquier cosa, ya que olvidarlo es la causa más común de confusión.

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

El servidor FastAPI más pequeño

FastAPI desempeña el papel que Express tiene en el mundo de JavaScript. Un ejemplo de “Hello World” se ve así:

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

Comiéncelo con uvicorn:

uvicorn main:app --reload

La bandera --reload es el equivalente de nodemon, ya que reinicia el servidor cada vez que se modifica un archivo. Una solicitud rápida prueba el endpoint:

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

Conectar Gmail

Ahora las cosas se vuelven reales, ya que el agente se comunica con una cuenta de Gmail real. Primero se procede con la parte de Google Cloud:

  1. Abra la consola de Google Cloud y cree un proyecto nuevo.
  2. Habilite tanto la API de Gmail como la API de Google Calendar.
  3. Cree credenciales OAuth del tipo Aplicación de escritorio.
  4. Descargue el archivo JSON y guárdelo como credentials.json.
  5. En la pantalla de consentimiento de OAuth, en la sección Público, agregue su propia dirección de Gmail como usuario de prueba.

Luego instale las bibliotecas de cliente de Google:

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

La herramienta de Gmail solo crea borradores y nunca los envía. Un agente no debería tener la facultad de enviar correos en su nombre sin pasar por un proceso de revisión.

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

La primera vez que lo ejecutas, una ventana del navegador te pedirá que inicies sesión con Google y te advertirá que la aplicación no está verificada. Elige Opciones avanzadas y luego ve a la aplicación. Esto es normal en software que aún está en desarrollo. Posteriormente, las credenciales se guardan en token.json, por lo que el inicio de sesión solo ocurre una vez.

Hay una regla de seguridad que merece ser enfatizada: nunca subas token.json ni credentials.json a git. El escaneo de secretos de GitHub suele bloquear ese tipo de envíos, pero es mejor no depender de ello. Añade los archivos a .gitignore de inmediato:

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

Conectar el Calendario de Google

El proceso es similar al de Gmail: autenticarse y luego llamar a la 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}"

El agente en sí

LangGraph conecta todas las partes entre sí. Instálalo junto con las integraciones del modelo:

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

El agente cuenta con cuatro herramientas, que son funciones ordinarias de Python que LangGraph puede llamar automáticamente una vez que cuenten con el decorador de herramienta:

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

Elegir llama3.1 en lugar de llama3.2

La llamada a herramientas requiere que el modelo genere JSON estructurado que LangGraph pueda analizar, y los modelos más pequeños tienen dificultades con eso. El modelo de 3B llama3.2 seguía devolviendo JSON en bruto en lugar de invocar realmente la herramienta, mientras que el modelo de 8B llama3.1 lo maneja de manera fiable. En el caso de los agentes, el tamaño del modelo es más importante que en una conversación normal.

Transmisión en tiempo real del razonamiento del agente

Una de las mejores características de LangGraph es la posibilidad de observar cómo un agente realiza una tarea paso a paso: qué herramienta eligió, qué resultado obtuvo y cómo llegó a una respuesta. Los eventos enviados por el servidor transmiten esa información al frontend en tiempo real. El código relevante se encuentra en 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")

En la interfaz, el usuario ve una secuencia como esta:

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

Ver al agente pensar en voz alta es una demostración persuasiva. Un script de prueba ejecutado contra el modelo local muestra el rastro completo:

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

El resultado utiliza correos electrónicos reales y eventos de calendario reales, con el agente eligiendo la herramienta adecuada sin ayuda, todo ello en una laptop sin conexión a Internet.

Lecciones importantes

  • El tamaño del modelo es más importante de lo esperado para los agentes. RAG funciona bien con modelos pequeños porque solo lee el contexto y lo resume. La llamada a las herramientas requiere JSON estructurado válido, y los modelos pequeños tienden a generar formatos erróneos. La diferencia entre un modelo de 8B y uno de 3B es enorme en este aspecto.
  • Mantenga los prompts cortos y explícitos. Un largo bloque de instrucciones detalladas suele ser ignorado por los modelos locales pequeños, mientras que unas seis líneas con una lista de viñetas funciona mucho mejor.
  • La opción de solo bocetos es la predeterminada adecuada. Hacer que create_draft sea la única acción de correo mantiene el control en manos humanas. Un agente autorizado a enviar correos de forma autónoma representa un riesgo, por lo que es necesario que los humanos revisen todo antes de que algo se envíe.
  • Los entornos virtuales castigan el olvido. Cada vez que aparece un ModuleNotFoundError, la solución suele ser activar nuevamente el venv.
  • Python es más sencillo de lo que parece para un desarrollador de JavaScript. No hay puntos y coma, ni llaves, y la estructura proviene del sangrado de espacios. Después de aproximadamente una semana de uso diario, el lenguaje pasa a un segundo plano y el problema ocupa el centro de atención.

Ejecutar el proyecto terminado

Los comandos a continuación clonan el proyecto terminado y lo inician:

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

¿A dónde ir a continuación?

  • Una herramienta de búsqueda en la web, para que el agente pueda buscar información en línea y resumirla.
  • Historial de conversaciones, ya que cada pregunta actualmente no tiene estado y la memoria de sesión permitiría continuaciones naturales.
  • Más herramientas, como notas, recordatorios, mensajes de Slack o problemas en GitHub. La fórmula nunca cambia: escribe una función, úsala con @tool y añádela al agente.
  • Despliegue en producción, con Docker, Claude en lugar del modelo local y un servicio como Railway o Render.

Los agentes no son mágicos. Son LLMs con funciones que pueden llamar, y una vez que eso funciona, todo el campo se abre. Es al construir algo donde ocurre el aprendizaje.

Leer más

← Todos los artículosEnviar el brief