Airflow MCP

Interactúa con Apache Airflow usando lenguaje natural para gestionar y monitorear tus flujos de trabajo de datos.

GitHub
10
Prueba este MCPPatrocinado

Documentación

⚠️ REPOSITORIO MOVIDO - YA NO SE MANTIENE AQUÍ

Este repositorio ha sido transferido a una nueva propiedad y ya no se mantiene activamente en esta ubicación.

🔄 Aviso de Migración

Este repositorio y todos los paquetes de código abierto asociados se han movido a una nueva organización de GitHub.

Nueva ubicación: https://github.com/ponderedw

📍 Qué significa esto

  • ✅ El desarrollo activo continúa en la nueva ubicación
  • ✅ Las últimas actualizaciones y versiones se publican allí
  • ✅ Los problemas y solicitudes de extracción deben enviarse al nuevo repositorio
  • ⚠️ Este repositorio ya no recibirá actualizaciones

🔗 Encuentra el Repositorio Actualizado

Visita https://github.com/ponderedw para:

  • Acceder a la última versión de este paquete
  • Reportar problemas o contribuir
  • Ver documentación actualizada
  • Obtener soporte de los mantenedores

Gracias por tu comprensión durante esta transición.

Airflow MCP

Este proyecto implementa un servidor MCP para Apache Airflow, permitiendo a los usuarios interactuar con su plataforma de orquestación usando lenguaje natural.

Con unos minutos de configuración, deberías poder usar Claude Desktop o cualquier LLM habilitado para MCP para hacer preguntas como:

  • "¿Qué DAGs tenemos en nuestro clúster de Airflow?"
  • "¿Cuál es nuestro último DAG fallido?"

¡Y más!

Acerca de MCP y Airflow MCP

El Protocolo de Contexto de Modelo (MCP) es un estándar abierto que crea conexiones seguras entre fuentes de datos y aplicaciones de IA. Este repositorio proporciona un servidor MCP personalizado para Apache Airflow que transforma cómo los equipos interactúan con su plataforma de orquestación a través del lenguaje natural.

🚀 Características

  • Consultar estados de pipelines mediante lenguaje natural
  • Solucionar fallos de DAG de manera eficiente
  • Obtener información completa de los DAGs
  • Activar DAGs según su estado
  • Monitorear resultados de ejecución
  • Analizar componentes y configuraciones de los DAGs

🛠️ Primeros Pasos

Requisitos Previos

Si ya tienes una instancia de Airflow y quieres usar nuestra imagen Docker preconstruida, solo necesitas:

  • Docker
  • Acceso a tu instancia de Apache Airflow
  • Acceso a un LLM (Claude, ChatGPT o AWS Bedrock)

Este repositorio también proporciona una configuración local para Apache Airflow, que puedes usar con fines de demostración.

También puedes compilar el servidor MCP desde el código fuente, como se detalla a continuación.

Inicio Rápido - Usando la Imagen Docker Preconstruida

Si tienes una instancia de Airflow y quieres usar nuestra imagen Docker preconstruida, simplemente sigue estos pasos:

Necesitarás configurar Claude Desktop para conectarse a tu instancia de Airflow. Si no has configurado Claude Desktop para usar MCP antes, te recomendamos seguir la documentación de Claude Desktop.

Estos son los pasos para configurar Claude Desktop y conectarlo a tu instancia de Airflow, usando nuestra imagen Docker preconstruida:

  • Abre Claude Desktop
  • Ve a Configuración → pestaña Desarrollador
  • Edita la configuración de MCP con:
{
   "mcpServers": {
         "airflow_mcp": {
            "command": "docker",
            "args": ["run", "-i", "--rm", "-e", "airflow_api_url", "-e", "airflow", "-e", "airflow", "hipposysai/airflow-mcp:latest"],
            "env": {
               "airflow_api_url": "http://host.docker.internal:8088/api/v1",
               "airflow_username": "airflow",
               "airflow_password": "airflow"
            }
         }
   }
}

Ejecutar MCP Localmente con Claude Desktop

  1. Clona este repositorio:

    git clone https://github.com/hipposys-ltd/airflow-mcp

  2. Si no tienes un entorno de Airflow en ejecución, inicia uno con:

    just airflow

    Esto iniciará una instancia de Airflow en el puerto 8088, con nombre de usuario airflow y contraseña airflow.

    Puedes acceder a Airflow en http://localhost:8088/ y ver múltiples DAGs configurados:

    Airflow DAGs.

    Estos DAGs tienen dependencias complejas, algunos se ejecutan según un horario y otros usan la funcionalidad de Datasets de Airflow.

  3. Configura Claude Desktop:

    Necesitarás configurar Claude Desktop para conectarse a tu instancia de Airflow. Si no has configurado Claude Desktop para usar MCP antes, te recomendamos seguir la documentación de Claude Desktop.

    Estos son los pasos para configurar Claude Desktop y conectarlo a tu instancia de Airflow:

    • Abre Claude Desktop
    • Ve a Configuración → pestaña Desarrollador
    • Edita la configuración de MCP con:
    {
   "mcpServers": {
       "airflow_mcp": {
           "command": "docker",
           "args": ["run", "-i", "--rm", "-e", "airflow_api_url", "-e", "airflow", "-e", "airflow", "hipposysai/airflow-mcp:latest"],
           "env": {
             "airflow_api_url": "http://host.docker.internal:8088/api/v1",
             "airflow_username": "airflow",
             "airflow_password": "airflow"
           }
       }
   }
}

  4. Prueba tu configuración preguntándole a Claude: "¿Qué DAGs tenemos en nuestro clúster de Airflow?"

Integración con LangChain

  1. Configura el entorno:

    cp template.env .env

  2. Configura tu modelo de LLM en .env:

    • Para AWS Bedrock: LLM_MODEL_ID=bedrock:...
    • Para Anthropic: LLM_MODEL_ID=anthropic:...
    • Para OpenAI: LLM_MODEL_ID=openai:...

  3. Agrega tus credenciales de API a .env:

    • Credenciales de AWS para Bedrock
    • ANTHROPIC_API_KEY para Claude
    • OPENAI_API_KEY para ChatGPT

  4. (Opcional) Conéctate a tu propio Airflow:

    airflow_api_url=your_airflow_api_url
airflow_username=your_airflow_username
airflow_password=your_airflow_password

  5. Inicia el proyecto:

    • Con Airflow incluido: just project
    • Con Airflow existente: just project_no_airflow

  6. Abre las interfaces web:

    just open_web_tabs

  7. Pruébalo preguntando "¿Cuántos DAGs fallaron hoy?" en la interfaz de chat

📝 Ejemplos de Uso

  • "¿Qué DAGs tenemos en nuestro clúster de Airflow?"
  • "Identifica todos los DAGs con estado fallido en su ejecución más reciente y activa una nueva ejecución para cada uno"
  • "¿Qué operadores usa el DAG transform_forecast_attendance?"
  • "¿El DAG transform_forecast_attendance se ha completado exitosamente alguna vez?"

Ejecutar MCP con LangChain

Puedes usar from langchain_mcp_adapters.client import MultiServerMCPClient para agregar nuestro Airflow MCP como una de tus herramientas en tu aplicación LangChain.

Opción 1: Contenedor Separado (Transporte SSE)

Si ejecutas nuestro Airflow MCP como un contenedor separado, usa transporte SSE:

mcp_host = os.environ.get('mcp_host', 'mcp_sse_server:8000')
mcps = {
    "AirflowMCP": {
        "url": f"http://{mcp_host}/sse",
        "transport": "sse",
        "headers": {"Authorization": f"""Bearer {
            os.environ.get('MCP_TOKEN')}"""}
    }
}

Opción 2: Servidor Integrado (Transporte STDIO)

Si quieres ejecutar nuestro servidor MCP como parte del código de LangChain, sin código externo, usa STDIO y asegúrate de instalar nuestra biblioteca primero (airflow-mcp-hipposys = "0.1.0a11"):

mcps = {
    "AirflowMCP":
    {
        'command': "python",
        'args': ["-m", "airflow_mcp_hipposys.mcp_airflow"],
        "transport": "stdio",
        'env': {k: v for k, v in {
            'AIRFLOW_ASSISTENT_AI_CONN': os.getenv(
                'AIRFLOW_ASSISTENT_AI_CONN'),
            'airflow_api_url': os.getenv('airflow_api_url'),
            'airflow_username': os.getenv('airflow_username'),
            'airflow_password': os.getenv('airflow_password'),
            'AIRFLOW_INSIGHTS_MODE':
                os.getenv('AIRFLOW_INSIGHTS_MODE'),
            'POST_MODE': os.getenv('POST_MODE'),
            'TRANSPORT_TYPE': 'stdio',
            '_AIRFLOW_WWW_USER_USERNAME':
                os.getenv('_AIRFLOW_WWW_USER_USERNAME'),
            '_AIRFLOW_WWW_USER_PASSWORD':
                os.getenv('_AIRFLOW_WWW_USER_PASSWORD')
        }.items() if v is not None}
    }
}

Usando las Herramientas

Luego, en ambos casos, pásalo a las herramientas:

client = MultiServerMCPClient(mcps)
tools = await client.get_tools()

🤝 Contribuciones

¡Invitamos entusiastamente a la comunidad a contribuir a esta iniciativa de código abierto! Ya sea que te interese:

  • Agregar nuevas características
  • Mejorar la documentación
  • Mejorar la compatibilidad con diferentes proveedores de LLM
  • Reportar errores
  • Sugerir mejoras

No dudes en enviar solicitudes de extracción o abrir problemas en nuestro repositorio de GitHub.

🔗 Enlaces