Construcción de un Agente de Operaciones Autónomo con Google ADK y n8n

De la Monitorización Pasiva a la Ejecución Agéntica

En la mayoría de organizaciones SaaS y plataformas de comercio electrónico, los fundadores y equipos técnicos operan como esclavos de sus propias métricas. Revisan manualmente paneles en busca de roturas de stock, pedidos bloqueados o reembolsos anómalos. Este modelo pull es ineficiente y no escala.

Invertir la lógica operativa implica construir una arquitectura event-driven donde la base de datos y los agentes de inteligencia artificial interactúan bidireccionalmente. Cuando un documento muta o un usuario interactúa, el agente analiza el contexto, evalúa el impacto del cambio, valida la consistencia transaccional y solicita autorización humana solo cuando la criticidad lo requiere.

Matriz de Decisión: Enfoques de Automatización y Operaciones

Implementación de Producción: Arquitectura Multi-Agente con Google ADK

A continuación se presenta la definición modular en Python del ecosistema de agentes, conectores MCP y herramientas de webhook hacia n8n:

import os
import requests
from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.tools import Tool, FunctionTool
from google.adk.mcp import MCPClientConnectionParams

load_dotenv()

# ==========================================
# 1. HERRAMIENTAS DETERMINISTAS (n8n Webhook)
# ==========================================
def generate_payment_link_tool(product_name: str, amount: float, customer_email: str) -> str:
    """Invoca el webhook transaccional de n8n para generar el enlace de pago y despachar el email."""
    webhook_url = os.getenv("N8N_PAYMENT_WEBHOOK_URL", "http://localhost:5678/webhook/payment-link")
    payload = {
        "product_name": product_name,
        "amount": amount,
        "customer_email": customer_email
    }
    
    try:
        response = requests.post(webhook_url, json=payload, timeout=10)
        response.raise_for_status()
        return f"Payment link dispatched successfully via n8n to {customer_email}."
    except requests.exceptions.RequestException as e:
        return f"Error triggering transactional workflow: {str(e)}"

# ==========================================
# 2. DEFINICIÓN DE SUB-AGENTES ESPECIALIZADOS
# ==========================================

# Agente de Inventario (Solo Lectura vía MCP)
inventory_agent = Agent(
    name="InventoryAgent",
    model="gemini-1.5-flash",
    description="Especialista en consulta de catálogo y verificación de stock.",
    instruction=(
        "Eres el agente de inventario. Tienes acceso de SOLO LECTURA a la base de datos "
        "a través del servidor MCP. Consulta precios exactos y existencias disponibles."
    ),
    mcp_servers=[
        MCPClientConnectionParams(
            command="docker",
            args=["run", "-i", "--rm", "mcp/mongodb-readonly"],
            env={"MONGO_URI": os.getenv("MONGO_READONLY_URI")}
        )
    ]
)

# Agente de Ventas y Mutaciones (Escritura controlada y Webhooks)
sales_agent = Agent(
    name="SalesAgent",
    model="gemini-1.5-flash",
    description="Responsable de emitir cobros y actualizar decrementos de stock.",
    instruction=(
        "Eres el agente de ventas. Valida el precio y el email del cliente antes de disparar el webhook "
        "de pago. Ejecuta la herramienta de actualización de stock una vez confirmada la intención de compra."
    ),
    tools=[
        FunctionTool(generate_payment_link_tool)
    ],
    mcp_servers=[
        MCPClientConnectionParams(
            command="docker",
            args=["run", "-i", "--rm", "mcp/mongodb-writable"],
            env={"MONGO_URI": os.getenv("MONGO_WRITABLE_URI")}
        )
    ]
)

# ==========================================
# 3. AGENTE RAÍZ (Operations Manager Router)
# ==========================================
root_operations_agent = Agent(
    name="OpManager",
    model="gemini-1.5-flash",
    description="Orquestador principal de operaciones de SaaS y E-commerce.",
    instruction=(
        "Eres el Operations Manager. Coordina el flujo de trabajo: "
        "1. Si el usuario consulta disponibilidad, delega a InventoryAgent. "
        "2. Cuando el usuario confirma una compra y provee su correo, delega a SalesAgent para emitir el enlace de pago y sincronizar stock."
    ),
    sub_agents=[inventory_agent, sales_agent]
)

if __name__ == "__main__":
    # Inicio del servidor agéntico ADK
    print("Iniciando servicio de operaciones autónomas con Google ADK...")

Patrones de Diseño y Robustez Arquitectónica

Para garantizar resiliencia en entornos de alta concurrencia empresarial, se deben contemplar tres pilares:

Guía de Implementación Paso a Paso

1. Levantar la infraestructura con Docker Compose: Despliega las instancias de base de datos (MongoDB o Firestore Emulator) y el contenedor de n8n autoalojado con persistencia de volúmenes.
2. Configurar el Workflow en n8n: Crea un flujo con un nodo Webhook (POST), seguido de la integración con Stripe API (Payment Link) y un nodo de correo Gmail / SMTP para responder con el status 200 OK.
3. Definir los Servidores MCP: Configura los adaptadores MCP para interactuar con tus colecciones de base de datos exponiendo esquemas JSON estrictos.
4. Instanciar el Agente con Google ADK: Ensambla la jerarquía de sub-agentes en Python conectando las credenciales de Vertex AI / Gemini.
5. Validación y Pruebas Human-in-the-Loop: Verifica en la interfaz web de ADK que las consultas no muten registros y que las operaciones de cobro se ejecuten de manera determinista.

Preguntas Frecuentes (FAQ)