MCP-Airflow-API

MCP-Airflow-API é um servidor MCP que utiliza o Protocolo de Contexto de Modelo (MCP) para transformar operações da API REST do Apache Airflow em ferramentas de linguagem natural. Este projeto oculta a complexidade das estruturas de API e permite o gerenciamento intuitivo de clusters Airflow por meio de comandos em linguagem natural.

GitHub
53
Experimente este MCPPatrocinado

Documentação

🚀 MCP-Airflow-API

Ferramenta Open Source Revolucionária para Gerenciar Apache Airflow com Linguagem Natural

License: MIT Python Docker Pulls BuyMeACoffee

Deploy to PyPI with tag PyPI PyPI - Downloads

Arquitetura & Internos (DeepWiki)

Ask DeepWiki

📋 Visão Geral

Você já imaginou como seria incrível poder gerenciar seus fluxos de trabalho do Apache Airflow usando linguagem natural em vez de chamadas complexas de API REST ou manipulações de interface web? MCP-Airflow-API é o projeto open source revolucionário que torna esse objetivo uma realidade.

MCP-Airflow-API Screenshot

🎯 O que é MCP-Airflow-API?

MCP-Airflow-API é um servidor MCP que utiliza o Model Context Protocol (MCP) para transformar operações da API REST do Apache Airflow em ferramentas de linguagem natural. Este projeto oculta a complexidade das estruturas de API e permite o gerenciamento intuitivo de clusters Airflow por meio de comandos em linguagem natural.

🆕 Suporte a Múltiplas Versões de API (NOVO!)

Agora suporta tanto a API v1 do Airflow (2.x) quanto a v2 (3.0+) com seleção dinâmica de versão via variável de ambiente:

  • API v1: Compatibilidade total com clusters Airflow 2.x (43 ferramentas) - Documentação
  • API v2: Recursos aprimorados para Airflow 3.0+, incluindo gerenciamento de assets para agendamento orientado a dados (45 ferramentas) - Documentação

Arquitetura Chave: Servidor MCP único com ferramentas comuns compartilhadas (43) mais ferramentas exclusivas de assets da v2 (2) - carrega dinamicamente o conjunto de ferramentas apropriado com base na variável de ambiente AIRFLOW_API_VERSION!

Abordagem tradicional (exemplo):

curl -X GET "http://localhost:8080/api/v1/dags?limit=100&offset=0" \
  -H "Authorization: Basic YWlyZmxvdzphaXJmbG93"

Abordagem MCP-Airflow-API (linguagem natural):

"Mostre-me os DAGs atualmente em execução"

🚀 Início Rápido

📝 Precisa de um cluster Airflow de teste? Use nosso projeto complementar Airflow-Docker-Compose com suporte para ambientes Airflow 2.x e Airflow 3.x!

Diagrama de Fluxo do Início Rápido/Tutorial

Flow Diagram of Quickstart/Tutorial

🎯 Recomendado: Docker Compose (Ambiente de Demonstração Completo)

Para avaliação e testes rápidos:

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API

# Configure your Airflow credentials
cp .env.example .env
# Edit .env with your Airflow API settings

# Start all services
docker-compose up -d

# Access OpenWebUI at http://localhost:3002/
# API documentation at http://localhost:8002/docs

Começando com OpenWebUI (Opção Docker)

📌 Nota: As instruções de configuração da interface web são baseadas no OpenWebUI v0.6.22. Os locais de menu e configurações podem diferir em versões mais recentes.

  1. Acesse http://localhost:3002/
  2. Faça login com a conta de administrador
  3. Vá para "Configurações" → "Ferramentas" no menu superior
  4. Adicione a URL da Ferramenta: http://localhost:8002/airflow-api
  5. Configure seu provedor de LLM (Ollama, OpenAI, etc.)

📦 Métodos de Instalação do Servidor MCP

Método 1: Instalação Direta do PyPI

uvx --python 3.12 mcp-airflow-api

Método 2: Integração com Cliente MCP Claude-Desktop

Acesso Local (modo stdio)

{
  "mcpServers": {
    "mcp-airflow-api": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v2",
        "AIRFLOW_API_BASE_URL": "http://localhost:8080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    }
  }
}git

Acesso Remoto (modo streamable-http sem autenticação)

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://localhost:8000/mcp"
    }
  }
}

Acesso Remoto (modo streamable-http com autenticação por token Bearer - Recomendado)

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://localhost:8000/mcp",
      "headers": {
        "Authorization": "Bearer your-secure-secret-key-here"
      }
    }
  }
}

Múltiplos Clusters Airflow com Versões Diferentes

{
  "mcpServers": {
    "airflow-2x-cluster": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v1",
        "AIRFLOW_API_BASE_URL": "http://localhost:38080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    },
    "airflow-3x-cluster": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v2",
        "AIRFLOW_API_BASE_URL": "http://localhost:48080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    }
  }
}

💡 Dica Profissional: Use os clusters de teste do Airflow-Docker-Compose para a configuração acima - eles rodam nas portas 38080 (2.x) e 48080 (3.x), respectivamente!

Método 3: Instalação para Desenvolvimento

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API
pip install -e .

# Run in stdio mode
python -m mcp_airflow_api

🌟 Principais Recursos

  1. Consultas em Linguagem Natural
    Não é necessário aprender sintaxe complexa de API. Basta perguntar como você falaria naturalmente:

    • "Quais DAGs estão em execução atualmente?"
    • "Mostre-me as tarefas com falha"
    • "Encontre DAGs contendo ETL"

  2. Capacidades Abrangentes de Monitoramento
    Monitoramento de status do cluster em tempo real:

    • Monitoramento de saúde do cluster
    • Análise de status e desempenho de DAGs
    • Rastreamento de logs de execução de tarefas
    • Gerenciamento de dados XCom

  3. Suporte Dinâmico a Versões de API
    Servidor MCP único que se adapta à sua versão do Airflow:

    • API v1: 43 ferramentas compartilhadas para compatibilidade com Airflow 2.x
    • API v2: 43 ferramentas compartilhadas + 2 ferramentas de gerenciamento de assets para Airflow 3.0+
    • Controle por Variável de Ambiente: Alterne versões instantaneamente com AIRFLOW_API_VERSION
    • Zero Mudanças de Configuração: Mesmos nomes de ferramentas, capacidades aprimoradas
    • Arquitetura Eficiente: Base de código comum compartilhada elimina duplicação

  4. Cobertura Abrangente de Ferramentas
    Cobre quase toda a funcionalidade da API do Airflow:

    • Gerenciamento de DAGs (disparar, pausar, retomar)
    • Monitoramento de instâncias de tarefas
    • Gerenciamento de pools e variáveis
    • Configuração de conexões
    • Consultas de configuração
    • Análise de logs de eventos

  5. Otimização para Ambientes Grandes
    Lida eficientemente com ambientes grandes com mais de 1000 DAGs:

    • Suporte a paginação inteligente
    • Opções avançadas de filtragem
    • Capacidades de processamento em lote

🛠️ Vantagens Técnicas

  • Utilizando o Model Context Protocol (MCP)
    MCP é um padrão aberto para conexões seguras entre aplicações de IA e fontes de dados, fornecendo:

    • Interface padronizada
    • Acesso seguro a dados
    • Arquitetura escalável

  • Suporte a Dois Modos de Transporte

    • Modo stdio: Integração direta com cliente MCP para ambientes locais
    • Modo streamable-http: Implantação baseada em HTTP para Docker e acesso remoto

    Controle por Variável de Ambiente:

    FASTMCP_TYPE=stdio          # Default: Direct MCP client mode
FASTMCP_TYPE=streamable-http # Docker/HTTP mode
FASTMCP_PORT=8000           # HTTP server port (Docker internal)

  • Cobertura Abrangente da API do Airflow
    Implementação completa das APIs REST oficiais do Airflow:

    • Suporte à API v1: Baseado na API REST do Airflow 2.x
    • Suporte à API v2: Baseado na API REST do Airflow 3.0+
    • Seleção Dinâmica de Versão: Alternância em tempo de execução entre versões de API
    • Paridade de Recursos: Cobertura completa de endpoints para ambas as versões

  • Suporte Completo a Docker
    Configuração completa do Docker Compose com 3 serviços separados:

    • Open WebUI: Interface web (porta 3002)
    • Servidor MCP: Ferramentas da API do Airflow (porta interna 8000, exposta via 18002)
    • Proxy MCPO: Provedor de endpoint de API REST (porta 8002)

Casos de Uso em Ação

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

⚙️ Configuração Avançada

Variáveis de Ambiente

# Required - Dynamic API Version Selection (NEW!)
# Single server supports both v1 and v2 - just change this variable!
AIRFLOW_API_VERSION=v1           # v1 for Airflow 2.x, v2 for Airflow 3.0+
AIRFLOW_API_BASE_URL=http://localhost:8080/api

# Test Cluster Connection Examples:
# For Airflow 2.x test cluster (from Airflow-Docker-Compose)
AIRFLOW_API_VERSION=v1
AIRFLOW_API_BASE_URL=http://localhost:38080/api

# For Airflow 3.x test cluster (from Airflow-Docker-Compose)  
AIRFLOW_API_VERSION=v2
AIRFLOW_API_BASE_URL=http://localhost:48080/api

# Authentication
AIRFLOW_API_USERNAME=airflow
AIRFLOW_API_PASSWORD=airflow

# Optional - MCP Server Configuration
MCP_LOG_LEVEL=INFO                   # DEBUG/INFO/WARNING/ERROR/CRITICAL
FASTMCP_TYPE=stdio                   # stdio/streamable-http
FASTMCP_PORT=8000                    # HTTP server port (Docker mode)

# Bearer Token Authentication for streamable-http mode
# Enable authentication (recommended for production)
# Default: false (when undefined, empty, or null)
# Values: true/false, 1/0, yes/no, on/off (case insensitive)
REMOTE_AUTH_ENABLE=false             # true/false
REMOTE_SECRET_KEY=your-secure-secret-key-here

Comparação de Versões de API

Documentação Oficial:

RecursoAPI v1 (Airflow 2.x)API v2 (Airflow 3.0+)
Total de Ferramentas43 ferramentas45 ferramentas
Ferramentas Compartilhadas43 (100%)43 (96%)
Ferramentas Exclusivas02 (Gerenciamento de Assets)
Operações Básicas de DAG✅✅ Aprimorado
Gerenciamento de Tarefas✅✅ Aprimorado
Gerenciamento de Conexões✅✅ Aprimorado
Gerenciamento de Pools✅✅ Aprimorado
Gerenciamento de Assets❌✅ Novo
Eventos de Assets❌✅ Novo
Agendamento Orientado a Dados❌✅ Novo
Avisos Aprimorados de DAG❌✅ Novo
Filtragem AvançadaBásica✅ Aprimorada

🔐 Segurança e Autenticação

Autenticação por Token Bearer

Para o modo streamable-http, este servidor MCP suporta autenticação por token Bearer para proteger o acesso remoto. Isso é especialmente importante ao executar o servidor em ambientes de produção.

Configuração

Habilitar Autenticação:

# In .env file
REMOTE_AUTH_ENABLE=true
REMOTE_SECRET_KEY=your-secure-secret-key-here

Ou via CLI:

python -m mcp_airflow_api --type streamable-http --auth-enable --secret-key your-secure-secret-key-here

Níveis de Segurança

  1. Modo stdio (Padrão): Acesso somente local, sem necessidade de autenticação
  2. streamable-http + REMOTE_AUTH_ENABLE=false: Acesso remoto sem autenticação ⚠️ NÃO RECOMENDADO para produção
  3. streamable-http + REMOTE_AUTH_ENABLE=true: Acesso remoto com autenticação por token Bearer ✅ RECOMENDADO para produção

Nota: REMOTE_AUTH_ENABLE assume o padrão false quando indefinido, vazio ou nulo. Os valores suportados são true/false, 1/0, yes/no, on/off (sem diferenciar maiúsculas de minúsculas).

Configuração do Cliente

Quando a autenticação está habilitada, os clientes MCP devem incluir o token Bearer no cabeçalho de Autorização:

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://your-server:8000/mcp",
      "headers": {
        "Authorization": "Bearer your-secure-secret-key-here"
      }
    }
  }
}

Boas Práticas de Segurança

  • Sempre habilite a autenticação ao usar o modo streamable-http em produção
  • Use chaves secretas fortes e geradas aleatoriamente (32+ caracteres recomendados)
  • Use HTTPS quando possível (configure proxy reverso com SSL/TLS)
  • Restrinja o acesso à rede usando firewalls ou políticas de rede
  • Rotacione chaves secretas regularmente para maior segurança
  • Monitore logs de acesso para tentativas de acesso não autorizado

Tratamento de Erros

Quando a autenticação falha, o servidor retorna:

  • 401 Não Autorizado para tokens ausentes ou inválidos
  • Mensagens de erro detalhadas em formato JSON para depuração

Configuração Personalizada do Docker Compose

version: '3.8'
services:
  mcp-server:
    build: 
      context: .
      dockerfile: Dockerfile.MCP-Server
    environment:
      - FASTMCP_PORT=8000
      - AIRFLOW_API_VERSION=v1
      - AIRFLOW_API_BASE_URL=http://your-airflow:8080/api
      - AIRFLOW_API_USERNAME=airflow
      - AIRFLOW_API_PASSWORD=airflow

Instalação para Desenvolvimento

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API
pip install -e .

# Run in stdio mode
python -m mcp_airflow_api

🧪 Implantação de Cluster Airflow de Teste

Para testes e desenvolvimento, use nosso projeto complementar Airflow-Docker-Compose que suporta ambientes Airflow 2.x e 3.x.

Configuração Rápida

  1. Clone o repositório do ambiente de teste: 
    git clone https://github.com/call518/Airflow-Docker-Compose.git
cd Airflow-Docker-Compose

Opção 1: Implantar Airflow 2.x (LTS)

Para testar a compatibilidade com API v1 com recursos estáveis de produção:

# Navigate to Airflow 2.x environment
cd airflow-2.x

# (Optional) Customize environment variables
cp .env.template .env
# Edit .env file as needed

# Deploy Airflow 2.x cluster
./run-airflow-cluster.sh

# Access Web UI
# URL: http://localhost:38080
# Username: airflow / Password: airflow

Detalhes do ambiente:

  • Imagem: apache/airflow:2.10.2
  • Porta: 38080 (configurável via AIRFLOW_WEBSERVER_PORT)
  • API: endpoints /api/v1/*
  • Autenticação: Autenticação Básica
  • Caso de uso: Pronto para produção, recursos estáveis

Opção 2: Implantar Airflow 3.x (Mais Recente)

Para testar a API v2 com os recursos mais recentes, incluindo gerenciamento de Assets:

# Navigate to Airflow 3.x environment  
cd airflow-3.x

# (Optional) Customize environment variables
cp .env.template .env
# Edit .env file as needed

# Deploy Airflow 3.x cluster
./run-airflow-cluster.sh

# Access API Server
# URL: http://localhost:48080
# Username: airflow / Password: airflow

Detalhes do ambiente:

  • Imagem: apache/airflow:3.0.6
  • Porta: 48080 (configurável via AIRFLOW_APISERVER_PORT)
  • API: endpoints /api/v2/* + gerenciamento de Assets
  • Autenticação: Token JWT (FabAuthManager)
  • Caso de uso: Desenvolvimento, teste de novos recursos

Opção 3: Implantar Ambas as Versões Simultaneamente

Para testes abrangentes em diferentes versões do Airflow:

# Start Airflow 2.x (port 38080)
cd airflow-2.x && ./run-airflow-cluster.sh

# Start Airflow 3.x (port 48080) 
cd ../airflow-3.x && ./run-airflow-cluster.sh

Principais Diferenças

RecursoAirflow 2.xAirflow 3.x
AutenticaçãoAutenticação BásicaTokens JWT (FabAuthManager)
Porta Padrão3808048080
Endpoints de API/api/v1/*/api/v2/*
Suporte a Assets❌ Limitado/Experimental✅ Suporte Completo
Pacotes de Provedoresprovidersdistributions
Estabilidade✅ Pronto para Produção🧪 Beta/Desenvolvimento

Limpeza

Para parar e limpar os ambientes de teste:

# For Airflow 2.x
cd airflow-2.x && ./cleanup-airflow-cluster.sh

# For Airflow 3.x
cd airflow-3.x && ./cleanup-airflow-cluster.sh

🌈 Arquitetura Preparada para o Futuro

  • Design escalável e estrutura modular para fácil adição de novos recursos
  • Protocolo em conformidade com padrões para integração com outras ferramentas
  • Operações nativas em nuvem e interface pronta para LLM
  • Processamento de consultas sensível ao contexto e capacidades de gerenciamento automatizado de fluxos de trabalho

🎯 Para Quem é Esta Ferramenta?

  • Engenheiros de Dados — Reduza o tempo de depuração, aumente a produtividade, minimize a curva de aprendizado
  • Engenheiros de DevOps — Automatize o monitoramento de infraestrutura, reduza o tempo de resposta a incidentes
  • Administradores de Sistemas — Gerenciamento amigável sem APIs complexas, monitoramento de status do cluster em tempo real

🚀 Contribuição Open Source e Comunidade

Repositório: https://github.com/call518/MCP-Airflow-API

Como Contribuir

  • Relatórios de bugs e sugestões de recursos
  • Melhorias na documentação
  • Contribuições de código

Por favor, considere dar uma estrela no projeto se você o achar útil.

🔮 Conclusão

MCP-Airflow-API muda o paradigma da engenharia de dados e do gerenciamento de fluxos de trabalho:
Não é necessário memorizar chamadas de API REST — basta perguntar em linguagem natural:

"Mostre-me o status dos trabalhos ETL atualmente em execução."

🏷️ Tags

#Apache-Airflow #MCP #ModelContextProtocol #DataEngineering #DevOps #WorkflowAutomation #NaturalLanguage #OpenSource #Python #Docker #AI-Integration

📚 Exemplos de Consultas e Casos de Uso

Esta seção fornece exemplos abrangentes de como usar as ferramentas do MCP-Airflow-API com consultas em linguagem natural.

Operações Básicas de DAG

  • list_dags: "Listar todos os DAGs com limite de 10 em formato de tabela." → Retorna até 10 DAGs
  • list_dags: "Listar todos os DAGs em formato de tabela." → Retorna todos os DAGs (AVISO: Requer muitos Tokens)
  • list_dags: "Mostrar próxima página de DAGs." → Use offset para paginação
  • list_dags: "Listar DAGs 21-40." → list_dags(limit=20, offset=20)
  • list_dags: "Filtrar DAGs cujo ID contém 'tutorial'." → list_dags(id_contains="etl")
  • list_dags: "Filtrar DAGs cujo nome de exibição contém 'tutorial'." → list_dags(name_contains="daily")
  • get_dags_detailed_batch: "Obter informações detalhadas para todos os DAGs com status de execução." → get_dags_detailed_batch(fetch_all=True)
  • get_dags_detailed_batch: "Obter detalhes para DAGs ativos e não pausados com execuções recentes." → get_dags_detailed_batch(is_active=True, is_paused=False)
  • get_dags_detailed_batch: "Obter informações detalhadas para DAGs contendo 'example' com histórico de execuções." → get_dags_detailed_batch(id_contains="example", limit=50)
  • running_dags: "Mostrar DAGs em execução."
  • failed_dags: "Mostrar DAGs com falha."
  • trigger_dag: "Disparar DAG 'example_complex'."
  • pause_dag: "Pausar DAG 'example_complex' em formato de tabela."
  • unpause_dag: "Despausar DAG 'example_complex' em formato de tabela."

Gerenciamento de Cluster e Saúde

  • get_health: "Verificar saúde do cluster Airflow."
  • get_version: "Obter informações da versão do Airflow."

Gerenciamento de Pools

  • list_pools: "Listar todos os pools."
  • list_pools: "Mostrar estatísticas de uso dos pools."
  • get_pool: "Obter detalhes do pool 'default_pool'."
  • get_pool: "Verificar utilização do pool."

Gerenciamento de Variáveis

  • list_variables: "Listar todas as variáveis."
  • list_variables: "Mostrar todas as variáveis do Airflow com seus valores."
  • get_variable: "Obter variável 'database_url'."
  • get_variable: "Mostrar o valor da variável 'api_key'."

Gerenciamento de Instâncias de Tarefas

  • list_task_instances_all: "Listar todas as instâncias de tarefas do DAG 'example_complex'."
  • list_task_instances_all: "Mostrar instâncias de tarefas em execução."
  • list_task_instances_all: "Mostrar instâncias de tarefas filtradas pelo pool 'default_pool'."
  • list_task_instances_all: "Listar instâncias de tarefas com duração maior que 300 segundos."
  • list_task_instances_all: "Mostrar instâncias de tarefas com falha da última semana."
  • list_task_instances_all: "Listar instâncias de tarefas com falha de ontem."
  • list_task_instances_all: "Mostrar instâncias de tarefas que iniciaram após as 9h de hoje."
  • list_task_instances_all: "Listar instâncias de tarefas dos últimos 3 dias com estado 'failed'."
  • get_task_instance_details: "Obter detalhes da tarefa 'data_processing' no DAG 'example_complex' execução 'scheduled__xxxxx'."
  • list_task_instances_batch: "Listar instâncias de tarefas com falha do último mês."
  • list_task_instances_batch: "Mostrar instâncias de tarefas em lote para múltiplos DAGs desta semana."
  • get_task_instance_extra_links: "Obter links extras para a tarefa 'data_processing' na execução mais recente."
  • get_task_instance_logs: "Recuperar logs da tarefa 'create_entry_gcs' tentativa número 2 do DAG 'example_complex'."

Gerenciamento de XCom

  • list_xcom_entries: "Listar entradas XCom da tarefa 'data_processing' no DAG 'example_complex' execução 'scheduled__xxxxx'."
  • list_xcom_entries: "Mostrar todas as entradas XCom da tarefa 'data_processing' na execução mais recente."
  • get_xcom_entry: "Obter entrada XCom com chave 'result' da tarefa 'data_processing' em execução específica."
  • get_xcom_entry: "Recuperar valor XCom da chave 'processed_count' da tarefa 'data_processing'."

Gerenciamento de Configuração

  • get_config: "Mostrar todas as seções e opções de configuração do Airflow." → Retorna configuração completa ou 403 se expose_config=False
  • list_config_sections: "Listar todas as seções de configuração com informações resumidas."
  • get_config_section: "Obter todas as configurações da seção 'core'." → get_config_section("core")
  • get_config_section: "Mostrar opções de configuração do webserver." → get_config_section("webserver")
  • search_config_options: "Encontrar todas as opções de configuração relacionadas a banco de dados." → search_config_options("database")
  • search_config_options: "Pesquisar configurações de timeout na configuração." → search_config_options("timeout")

Importante: As ferramentas de configuração exigem expose_config = True na seção [webserver] do airflow.cfg. Até usuários administradores recebem erros 403 se isso estiver desabilitado.

Análise e Monitoramento de DAGs

  • get_dag: "Obter detalhes do DAG 'example_complex'."
  • get_dags_detailed_batch: "Obter detalhes abrangentes de todos os DAGs com histórico de execução." → get_dags_detailed_batch(fetch_all=True)
  • get_dags_detailed_batch: "Obter detalhes dos DAGs ativos com informações da execução mais recente." → get_dags_detailed_batch(is_active=True)
  • get_dags_detailed_batch: "Obter informações detalhadas dos DAGs ETL com dados de execução recentes." → get_dags_detailed_batch(id_contains="etl")

Nota: get_dags_detailed_batch retorna cada DAG com detalhes de configuração (de get_dag()) e um campo latest_dag_run contendo as informações mais recentes de execução (run_id, state, execution_date, start_date, end_date, etc.).

  • dag_graph: "Mostrar grafo de tarefas do DAG 'example_complex'."
  • list_tasks: "Listar todas as tarefas do DAG 'example_complex'."
  • dag_code: "Obter código-fonte do DAG 'example_complex'."
  • list_event_logs: "Listar logs de eventos do DAG 'example_complex'."
  • list_event_logs: "Mostrar logs de eventos com ID de ontem para todos os DAGs."
  • get_event_log: "Obter entrada de log de evento com ID 12345."
  • all_dag_event_summary: "Mostrar resumo de contagem de eventos para todos os DAGs."
  • list_import_errors: "Listar erros de importação com ID."
  • get_import_error: "Obter erro de importação com ID 67890."
  • all_dag_import_summary: "Mostrar resumo de erros de importação para todos os DAGs."
  • dag_run_duration: "Obter estatísticas de duração de execução do DAG 'example_complex'."
  • dag_task_duration: "Mostrar execução mais recente do DAG 'example_complex'."
  • dag_task_duration: "Mostrar durações de tarefas da execução mais recente de 'manual__xxxxx'."
  • dag_calendar: "Obter informações de calendário do DAG 'example_complex' do último mês."
  • dag_calendar: "Mostrar agendamento do DAG 'example_complex' desta semana."

Exemplos de Cálculo de Datas

As ferramentas baseiam automaticamente os cálculos de datas relativas na data/hora atual do servidor:

Entrada do UsuárioMétodo de CálculoFormato de Exemplo
"ontem"data_atual - 1 diaYYYY-MM-DD (1 dia antes da atual)
"semana passada"data_atual - 7 dias até data_atual - 1 diaYYYY-MM-DD até YYYY-MM-DD (intervalo de 7 dias)
"últimos 3 dias"data_atual - 3 dias até data_atualYYYY-MM-DD até YYYY-MM-DD (intervalo de 3 dias)
"esta manhã"data_atual 00:00 até 12:00Formato YYYY-MM-DDTHH:mm:ssZ

O servidor sempre usa sua data/hora atual para esses cálculos.

Gerenciamento de Assets (Somente API v2)

Disponível apenas quando AIRFLOW_API_VERSION=v2 (Airflow 3.0+):

  • list_assets: "Mostrar todos os assets registrados no sistema." → Lista todos os assets de dados para agendamento orientado a dados
  • list_assets: "Encontrar assets com URI contendo 's3://data-lake'." → list_assets(uri_pattern="s3://data-lake")
  • list_asset_events: "Mostrar eventos recentes de assets." → Lista quando os assets foram criados ou atualizados
  • list_asset_events: "Mostrar eventos de assets para URI específica." → list_asset_events(asset_uri="s3://bucket/file.csv")
  • list_asset_events: "Encontrar eventos produzidos por DAGs ETL." → list_asset_events(source_dag_id="etl_pipeline")

Exemplos de Agendamento Orientado a Dados:

  • "Mostre-me quais assets disparam o DAG customer_analysis."
  • "Liste todos os assets criados pelo DAG data_ingestion esta semana."
  • "Encontre assets que não foram atualizados recentemente."
  • "Mostre a linhagem de dados do nosso pipeline de treinamento de ML."

Contribuindo

🤝 Tem ideias? Encontrou bugs? Quer adicionar recursos legais?

Estamos sempre animados em receber novos contribuidores! Seja corrigindo um erro de digitação, adicionando uma nova ferramenta de monitoramento ou melhorando a documentação - cada contribuição torna este projeto melhor.

Formas de contribuir:

  • 🐛 Reportar problemas ou bugs
  • 💡 Sugerir novos recursos de monitoramento do Airflow
  • 📝 Melhorar a documentação
  • 🚀 Enviar pull requests
  • ⭐ Dar estrela no repositório se você achar útil!

Dica profissional: O código foi projetado para ser super amigável para adicionar novas ferramentas. Confira as funções @mcp.tool() existentes em airflow_api.py.

🛠️ Adicionando Ferramentas Personalizadas (Avançado)

Este servidor MCP foi projetado para fácil extensibilidade. Depois de explorar os recursos principais e o Quickstart, você pode adicionar suas próprias ferramentas personalizadas da seguinte forma:

Guia Passo a Passo

1. Adicionar Funções Auxiliares (Opcional)

Adicione funções de dados reutilizáveis em src/mcp_airflow_api/functions.py:

async def get_your_custom_data(target_resource: str = None) -> List[Dict[str, Any]]:
  """Your custom data retrieval function."""
  # Example implementation - adapt to your service
  data_source = await get_data_connection(target_resource)
  results = await fetch_data_from_source(
    source=data_source,
    filters=your_conditions,
    aggregations=["count", "sum", "avg"],
    sorting=["count DESC", "timestamp ASC"]
  )
  return results

2. Criar Sua Ferramenta MCP

Adicione sua função de ferramenta em src/mcp_airflow_api/airflow_api.py:

@mcp.tool()
async def get_your_custom_analysis(limit: int = 50, target_name: Optional[str] = None) -> str:
  """
  [Tool Purpose]: Brief description of what your tool does
    
  [Exact Functionality]:
  - Feature 1: Data aggregation and analysis
  - Feature 2: Resource monitoring and insights
  - Feature 3: Performance metrics and reporting
    
  [Required Use Cases]:
  - When user asks "your specific analysis request"
  - Your business-specific monitoring needs
    
  Args:
    limit: Maximum results (1-100)
    target_name: Target resource/service name
    
  Returns:
    Formatted analysis results
  """
  try:
    limit = max(1, min(limit, 100))  # Always validate input
    results = await get_your_custom_data(target_resource=target_name)
    if results:
      results = results[:limit]
    return format_table_data(results, f"Custom Analysis (Top {len(results)})")
  except Exception as e:
    logger.error(f"Failed to get custom analysis: {e}")
    return f"Error: {str(e)}"

3. Atualizar Imports (Se Necessário)

Adicione sua função auxiliar aos imports em src/mcp_airflow_api/airflow_api.py:

from .functions import (
  # ...existing imports...
  get_your_custom_data,  # Add your new function
)

4. Atualizar o Template de Prompt (Recomendado)

Adicione a descrição da sua ferramenta em src/mcp_airflow_api/prompt_template.md para melhor reconhecimento de linguagem natural:

### **Your Custom Analysis Tool**

### X. **get_your_custom_analysis**
**Purpose**: Brief description of what your tool does
**Usage**: "Show me your custom analysis" or "Get custom analysis for database_name"
**Features**: Data aggregation, resource monitoring, performance metrics
**Required**: `target_name` parameter for specific resource analysis

5. Testar Sua Ferramenta

# Local testing
./scripts/run-mcp-inspector-local.sh

# Or with Docker
docker-compose up -d
docker-compose logs -f mcp-server

# Test with natural language:
# "Show me your custom analysis"
# "Get custom analysis for target_name"

Pronto! Sua ferramenta personalizada está pronta para uso com consultas em linguagem natural.

Licença

Use, modifique e distribua livremente sob a Licença MIT.