Airflow MCP Server

Servidor MCP para Airflow

Documentação

mcp-server-apache-airflow

PyPI - Downloads

Uma implementação de servidor Model Context Protocol (MCP) para Apache Airflow, permitindo integração perfeita com clientes MCP. Este projeto fornece uma forma padronizada de interagir com o Apache Airflow por meio do Model Context Protocol.

Sobre

Este projeto implementa um servidor Model Context Protocol que encapsula a API REST do Apache Airflow, permitindo que clientes MCP interajam com o Airflow de forma padronizada. Ele utiliza a biblioteca oficial do cliente Apache Airflow para garantir compatibilidade e manutenibilidade.

Status de Implementação de Recursos

Recurso	Caminho da API	Status
Gerenciamento de DAGs
Listar DAGs	`/api/v1/dags`	✅
Obter detalhes do DAG	`/api/v1/dags/{dag_id}`	✅
Pausar DAG	`/api/v1/dags/{dag_id}`	✅
Despausar DAG	`/api/v1/dags/{dag_id}`	✅
Atualizar DAG	`/api/v1/dags/{dag_id}`	✅
Excluir DAG	`/api/v1/dags/{dag_id}`	✅
Obter fonte do DAG	`/api/v1/dagSources/{file_token}`	✅
Aplicar patch em múltiplos DAGs	`/api/v1/dags`	✅
Reanalisar arquivo DAG	`/api/v1/dagSources/{file_token}/reparse`	✅
Execuções de DAG
Listar execuções de DAG	`/api/v1/dags/{dag_id}/dagRuns`	✅
Criar execução de DAG	`/api/v1/dags/{dag_id}/dagRuns`	✅
Obter detalhes da execução de DAG	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
Atualizar execução de DAG	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
Excluir execução de DAG	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
Obter lote de execuções de DAG	`/api/v1/dags/~/dagRuns/list`	✅
Limpar execução de DAG	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clear`	✅
Definir nota da execução de DAG	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/setNote`	✅
Obter eventos de dataset upstream	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/upstreamDatasetEvents`	✅
Tarefas
Listar tarefas do DAG	`/api/v1/dags/{dag_id}/tasks`	✅
Obter detalhes da tarefa	`/api/v1/dags/{dag_id}/tasks/{task_id}`	✅
Obter instância de tarefa	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}`	✅
Listar instâncias de tarefa	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances`	✅
Atualizar instância de tarefa	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}`	✅
Obter log da instância de tarefa	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/logs/{task_try_number}`	✅
Limpar instâncias de tarefa	`/api/v1/dags/{dag_id}/clearTaskInstances`	✅
Definir estado das instâncias de tarefa	`/api/v1/dags/{dag_id}/updateTaskInstancesState`	✅
Listar tentativas de instância de tarefa	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/tries`	✅
Variáveis
Listar variáveis	`/api/v1/variables`	✅
Criar variável	`/api/v1/variables`	✅
Obter variável	`/api/v1/variables/{variable_key}`	✅
Atualizar variável	`/api/v1/variables/{variable_key}`	✅
Excluir variável	`/api/v1/variables/{variable_key}`	✅
Conexões
Listar conexões	`/api/v1/connections`	✅
Criar conexão	`/api/v1/connections`	✅
Obter conexão	`/api/v1/connections/{connection_id}`	✅
Atualizar conexão	`/api/v1/connections/{connection_id}`	✅
Excluir conexão	`/api/v1/connections/{connection_id}`	✅
Testar conexão	`/api/v1/connections/test`	✅
Pools
Listar pools	`/api/v1/pools`	✅
Criar pool	`/api/v1/pools`	✅
Obter pool	`/api/v1/pools/{pool_name}`	✅
Atualizar pool	`/api/v1/pools/{pool_name}`	✅
Excluir pool	`/api/v1/pools/{pool_name}`	✅
XComs
Listar XComs	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries`	✅
Obter entrada XCom	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key}`	✅
Datasets
Listar datasets	`/api/v1/datasets`	✅
Obter dataset	`/api/v1/datasets/{uri}`	✅
Obter eventos de dataset	`/api/v1/datasetEvents`	✅
Criar evento de dataset	`/api/v1/datasetEvents`	✅
Obter evento de dataset enfileirado do DAG	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}`	✅
Obter eventos de dataset enfileirados do DAG	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents`	✅
Excluir evento de dataset enfileirado do DAG	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}`	✅
Excluir eventos de dataset enfileirados do DAG	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents`	✅
Obter eventos de dataset enfileirados	`/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents`	✅
Excluir eventos de dataset enfileirados	`/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents`	✅
Monitoramento
Obter status de saúde	`/api/v1/health`	✅
Estatísticas de DAG
Obter estatísticas de DAG	`/api/v1/dags/statistics`	✅
Configuração
Obter configuração	`/api/v1/config`	✅
Plugins
Obter plugins	`/api/v1/plugins`	✅
Provedores
Listar provedores	`/api/v1/providers`	✅
Logs de eventos
Listar logs de eventos	`/api/v1/eventLogs`	✅
Obter log de evento	`/api/v1/eventLogs/{event_log_id}`	✅
Sistema
Obter erros de importação	`/api/v1/importErrors`	✅
Obter detalhes do erro de importação	`/api/v1/importErrors/{import_error_id}`	✅
Obter status de saúde	`/api/v1/health`	✅
Obter versão	`/api/v1/version`	✅

Configuração

Dependências

Este projeto depende da biblioteca oficial do cliente Apache Airflow (apache-airflow-client). Ela será instalada automaticamente quando você instalar este pacote.

Variáveis de Ambiente

Defina as seguintes variáveis de ambiente:

AIRFLOW_HOST=<your-airflow-host>        # Optional, defaults to http://localhost:8080
AIRFLOW_API_VERSION=v1                  # Optional, defaults to v1
READ_ONLY=true                          # Optional, enables read-only mode (true/false, defaults to false)

Autenticação

Escolha um dos seguintes métodos de autenticação:

Autenticação Básica (padrão):

AIRFLOW_USERNAME=<your-airflow-username>
AIRFLOW_PASSWORD=<your-airflow-password>

Autenticação com Token JWT:

AIRFLOW_JWT_TOKEN=<your-jwt-token>

Para obter um token JWT, você pode usar o endpoint de autenticação do Airflow:

ENDPOINT_URL="http://localhost:8080"  # Replace with your Airflow endpoint
curl -X 'POST' \
  "${ENDPOINT_URL}/auth/token" \
  -H 'Content-Type: application/json' \
  -d '{ "username": "<your-username>", "password": "<your-password>" }'

Nota: Se tanto o token JWT quanto as credenciais de autenticação básica forem fornecidos, o token JWT terá precedência.

Uso com Claude Desktop

Adicione ao seu claude_desktop_config.json:

Autenticação Básica:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password"
      }
    }
  }
}

Autenticação com Token JWT:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

Para modo somente leitura (recomendado por segurança):

Autenticação Básica:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password",
        "READ_ONLY": "true"
      }
    }
  }
}

Autenticação com Token JWT:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow", "--read-only"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

Configuração alternativa usando uv:

Autenticação Básica:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password"
      }
    }
  }
}

Autenticação com Token JWT:

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

Substitua /path/to/mcp-server-apache-airflow pelo caminho real onde você clonou o repositório.

Selecionando os grupos de API

Você pode selecionar os grupos de API que deseja usar definindo a flag --apis.

uv run mcp-server-apache-airflow --apis dag --apis dagrun

O padrão é usar todas as APIs.

Os valores permitidos são:

config
connections
dag
dagrun
dagstats
dataset
eventlog
importerror
monitoring
plugin
pool
provider
taskinstance
variable
xcom

Modo Somente Leitura

Você pode executar o servidor em modo somente leitura usando a flag --read-only ou definindo a variável de ambiente READ_ONLY=true. Isso exporá apenas ferramentas que realizam operações de leitura (requisições GET) e excluirá quaisquer ferramentas que criem, atualizem ou excluam recursos.

Usando a flag de linha de comando:

uv run mcp-server-apache-airflow --read-only

Usando a variável de ambiente:

READ_ONLY=true uv run mcp-server-apache-airflow

No modo somente leitura, o servidor exporá apenas ferramentas como:

Listar DAGs, execuções de DAG, tarefas, variáveis, conexões, etc.
Obter detalhes de recursos específicos
Ler configurações e informações de monitoramento
Testar conexões (não destrutivo)

Operações de escrita, como criar, atualizar, excluir DAGs, variáveis, conexões, acionar execuções de DAG, etc., não estarão disponíveis no modo somente leitura.

Você pode combinar o modo somente leitura com a seleção de grupos de API:

uv run mcp-server-apache-airflow --read-only --apis dag --apis variable

Execução Manual

Você também pode executar o servidor manualmente:

make run

make run aceita as seguintes opções:

Opções:

--port: Porta para escutar SSE (padrão: 8000)
--transport: Tipo de transporte (stdio/sse/http, padrão: stdio)

Ou você pode executar o servidor SSE diretamente, que aceita os mesmos parâmetros:

make run-sse

Além disso, você pode iniciar o serviço diretamente usando uv como no seguinte comando:

uv run src --transport http --port 8080

Instalando via Smithery

Para instalar o Apache Airflow MCP Server para Claude Desktop automaticamente via Smithery:

npx -y @smithery/cli install @yangkyeongmo/mcp-server-apache-airflow --client claude

Desenvolvimento

Configurando o Ambiente de Desenvolvimento

Clone o repositório:

git clone https://github.com/yangkyeongmo/mcp-server-apache-airflow.git
cd mcp-server-apache-airflow

Instale as dependências de desenvolvimento:

uv sync --dev

Crie um arquivo .env para variáveis de ambiente (opcional para desenvolvimento):

touch .env

Nota: Nenhuma variável de ambiente é necessária para executar os testes. O AIRFLOW_HOST tem como padrão http://localhost:8080 para fins de desenvolvimento e teste.

Executando Testes

O projeto usa pytest para testes com os seguintes comandos disponíveis:

# Run all tests
make test

Qualidade de Código

# Run linting
make lint

# Run code formatting
make format

Integração Contínua

O projeto inclui um fluxo de trabalho do GitHub Actions (.github/workflows/test.yml) que automaticamente:

Executa testes em Python 3.10, 3.11 e 3.12
Executa verificações de lint usando ruff
Executa em cada push e pull request para o branch main

O pipeline de CI garante a qualidade do código e a compatibilidade entre as versões suportadas do Python antes que qualquer alteração seja mesclada.

Contribuindo

Contribuições são bem-vindas! Sinta-se à vontade para enviar um Pull Request.

O pacote é implantado automaticamente no PyPI quando project.version é atualizado em pyproject.toml. Siga o semver para versionamento.

Inclua a atualização de versão no PR para aplicar as alterações na lógica principal.

Licença

Licença MIT