Airflow MCP

Interaja com o Apache Airflow usando linguagem natural para gerenciar e monitorar seus fluxos de dados.

GitHub
10
Experimente este MCPPatrocinado

Documentação

⚠️ REPOSITÓRIO MOVIDO - NÃO É MAIS MANTIDO AQUI

Este repositório foi transferido para nova propriedade e não é mais mantido ativamente neste local.

🔄 Aviso de Migração

Este repositório e todos os pacotes de código aberto associados foram movidos para uma nova organização no GitHub.

Novo Local: https://github.com/ponderedw

📍 O Que Isso Significa

  • ✅ Desenvolvimento ativo continua no novo local
  • ✅ Atualizações mais recentes e lançamentos são publicados lá
  • ✅ Issues e pull requests devem ser submetidos ao novo repositório
  • ⚠️ Este repositório não receberá mais atualizações

🔗 Encontre o Repositório Atualizado

Por favor, visite https://github.com/ponderedw para:

  • Acessar a versão mais recente deste pacote
  • Reportar problemas ou contribuir
  • Visualizar documentação atualizada
  • Obter suporte dos mantenedores

Obrigado pela sua compreensão durante esta transição.

Airflow MCP

Este projeto implementa um servidor MCP para Apache Airflow, permitindo que os usuários interajam com sua plataforma de orquestração usando linguagem natural.

Com alguns minutos de configuração, você deve ser capaz de usar o Claude Desktop ou qualquer LLM habilitado para MCP para fazer perguntas como:

  • "Quais DAGs temos em nosso cluster Airflow?"
  • "Qual é o nosso DAG com falha mais recente?"

E mais!

Sobre MCP e Airflow MCP

O Model Context Protocol (MCP) é um padrão aberto que cria conexões seguras entre fontes de dados e aplicações de IA. Este repositório fornece um servidor MCP personalizado para Apache Airflow que transforma a forma como as equipes interagem com sua plataforma de orquestração por meio de linguagem natural.

🚀 Recursos

  • Consultar status de pipelines por meio de linguagem natural
  • Solucionar falhas de DAG de forma eficiente
  • Recuperar informações abrangentes de DAG
  • Acionar DAGs com base em seu status
  • Monitorar resultados de execução
  • Analisar componentes e configurações de DAG

🛠️ Começando

Pré-requisitos

Se você já tem uma instância do Airflow e deseja usar nossa imagem Docker pré-construída, você só precisa de:

  • Docker
  • Acesso à sua instância do Apache Airflow
  • Acesso a LLM (Claude, ChatGPT ou AWS Bedrock)

Este repositório também fornece uma configuração local para Apache Airflow, que você pode usar para fins de demonstração.

Você também pode compilar o servidor MCP a partir do código-fonte, detalhado abaixo.

Início Rápido - Usando a Imagem Docker Pré-construída

Se você tem uma instância do Airflow e deseja usar nossa imagem Docker pré-construída, basta seguir estes passos:

Você precisará configurar o Claude Desktop para conectar-se à sua instância do Airflow. Se você ainda não configurou o Claude Desktop para uso com MCP, recomendamos seguir a documentação do Claude Desktop.

Aqui estão os passos para configurar o Claude Desktop para conectar-se à sua instância do Airflow, usando nossa imagem Docker pré-construída:

  • Abra o Claude Desktop
  • Vá para Configurações → aba Desenvolvedor
  • Edite a configuração do MCP com:
{
   "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"
            }
         }
   }
}

Executando MCP Localmente com Claude Desktop

  1. Clone este repositório:

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

  2. Se você não tem um ambiente Airflow em execução, inicie um com:

    just airflow

    Isso iniciará uma instância do Airflow na porta 8088, com usuário airflow e senha airflow.

    Você pode acessar o Airflow em http://localhost:8088/ e ver vários DAGs configurados:

    Airflow DAGs.

    Esses DAGs têm dependências complexas, alguns executando em um agendamento e alguns usando a funcionalidade de Dataset do Airflow.

  3. Configure o Claude Desktop:

    Você precisará configurar o Claude Desktop para conectar-se à sua instância do Airflow. Se você ainda não configurou o Claude Desktop para uso com MCP, recomendamos seguir a documentação do Claude Desktop.

    Aqui estão os passos para configurar o Claude Desktop para conectar-se à sua instância do Airflow:

    • Abra o Claude Desktop
    • Vá para Configurações → aba Desenvolvedor
    • Edite a configuração do MCP com:
    {
   "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. Teste sua configuração perguntando ao Claude: "Quais DAGs temos em nosso cluster Airflow?"

Integrando com LangChain

  1. Configure o ambiente:

    cp template.env .env

  2. Configure seu modelo de LLM em .env:

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

  3. Adicione suas credenciais de API em .env:

    • Credenciais AWS para Bedrock
    • ANTHROPIC_API_KEY para Claude
    • OPENAI_API_KEY para ChatGPT

  4. (Opcional) Conecte-se ao seu próprio Airflow:

    airflow_api_url=your_airflow_api_url
airflow_username=your_airflow_username
airflow_password=your_airflow_password

  5. Inicie o projeto:

    • Com Airflow incluído: just project
    • Com Airflow existente: just project_no_airflow

  6. Abra as interfaces web:

    just open_web_tabs

  7. Experimente perguntando "Quantos DAGs falharam hoje?" na interface de chat

📝 Exemplo de Uso

  • "Quais DAGs temos em nosso cluster Airflow?"
  • "Identifique todos os DAGs com status de falha em sua execução mais recente e acione uma nova execução para cada um"
  • "Quais operadores são usados pelo DAG transform_forecast_attendance?"
  • "O DAG transform_forecast_attendance já foi concluído com sucesso?"

Executando MCP com LangChain

Você pode usar from langchain_mcp_adapters.client import MultiServerMCPClient para adicionar nosso Airflow MCP como uma de suas ferramentas no seu aplicativo LangChain.

Opção 1: Contêiner Separado (Transporte SSE)

Se você executar nosso Airflow MCP como um contêiner separado, use o 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')}"""}
    }
}

Opção 2: Servidor Embutido (Transporte STDIO)

Se você quiser executar nosso servidor MCP como parte do código LangChain, sem nenhum código externo, use STDIO e certifique-se de instalar nossa biblioteca primeiro (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 as Ferramentas

Então, em ambos os casos, passe-o para as ferramentas:

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

🤝 Contribuindo

Convidamos entusiasticamente a comunidade a contribuir com esta iniciativa de código aberto! Se você tem interesse em:

  • Adicionar novos recursos
  • Melhorar a documentação
  • Aprimorar a compatibilidade com diferentes provedores de LLM
  • Reportar bugs
  • Sugerir melhorias

Sinta-se à vontade para enviar pull requests ou abrir issues em nosso repositório GitHub.

🔗 Links