Skip to main content
Glama
MCP Server Airflow Token

MCP Server Airflow Token

mcp-server-airflow-token

A Model Context Protocol (MCP) server for Apache Airflow with Bearer token authentication support, enabling seamless integration with Astronomer Cloud and standalone Airflow instances.

Based on mcp-server-apache-airflow by Gyeongmo Nathan Yang

This fork enhances the original MCP server with Bearer token authentication support, making it compatible with Astronomer Cloud and other token-based Airflow deployments.

Key Enhancements

  • ✅ Bearer Token Authentication - Primary authentication method for modern Airflow deployments

  • ✅ Astronomer Cloud Compatible - Works seamlessly with Astronomer's managed Airflow

  • ✅ Backward Compatible - Still supports username/password authentication

  • ✅ Enhanced URL Handling - Correctly handles deployment paths like /deployment-id

Related MCP server: Messages MCP Server

About

This project implements a Model Context Protocol server that wraps Apache Airflow's REST API, allowing MCP clients to interact with Airflow in a standardized way. It uses the official Apache Airflow client library to ensure compatibility and maintainability.

Feature Implementation Status

Feature

API Path

Status

DAG Management

List DAGs

/api/v1/dags

✅

Get DAG Details

/api/v1/dags/{dag_id}

✅

Pause DAG

/api/v1/dags/{dag_id}

✅

Unpause DAG

/api/v1/dags/{dag_id}

✅

Update DAG

/api/v1/dags/{dag_id}

✅

Delete DAG

/api/v1/dags/{dag_id}

✅

Get DAG Source

/api/v1/dagSources/{file_token}

✅

Patch Multiple DAGs

/api/v1/dags

✅

Reparse DAG File

/api/v1/dagSources/{file_token}/reparse

✅

DAG Runs

List DAG Runs

/api/v1/dags/{dag_id}/dagRuns

✅

Create DAG Run

/api/v1/dags/{dag_id}/dagRuns

✅

Get DAG Run Details

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}

✅

Update DAG Run

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}

✅

Delete DAG Run

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}

✅

Get DAG Runs Batch

/api/v1/dags/~/dagRuns/list

✅

Clear DAG Run

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clear

✅

Set DAG Run Note

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/setNote

✅

Get Upstream Dataset Events

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/upstreamDatasetEvents

✅

Tasks

List DAG Tasks

/api/v1/dags/{dag_id}/tasks

✅

Get Task Details

/api/v1/dags/{dag_id}/tasks/{task_id}

✅

Get Task Instance

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}

✅

List Task Instances

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances

✅

Update Task Instance

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}

✅

Clear Task Instances

/api/v1/dags/{dag_id}/clearTaskInstances

✅

Set Task Instances State

/api/v1/dags/{dag_id}/updateTaskInstancesState

✅

Variables

List Variables

/api/v1/variables

✅

Create Variable

/api/v1/variables

✅

Get Variable

/api/v1/variables/{variable_key}

✅

Update Variable

/api/v1/variables/{variable_key}

✅

Delete Variable

/api/v1/variables/{variable_key}

✅

Connections

List Connections

/api/v1/connections

✅

Create Connection

/api/v1/connections

✅

Get Connection

/api/v1/connections/{connection_id}

✅

Update Connection

/api/v1/connections/{connection_id}

✅

Delete Connection

/api/v1/connections/{connection_id}

✅

Test Connection

/api/v1/connections/test

✅

Pools

List Pools

/api/v1/pools

✅

Create Pool

/api/v1/pools

✅

Get Pool

/api/v1/pools/{pool_name}

✅

Update Pool

/api/v1/pools/{pool_name}

✅

Delete Pool

/api/v1/pools/{pool_name}

✅

XComs

List XComs

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries

✅

Get XCom Entry

/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key}

✅

Datasets

List Datasets

/api/v1/datasets

✅

Get Dataset

/api/v1/datasets/{uri}

✅

Get Dataset Events

/api/v1/datasetEvents

✅

Create Dataset Event

/api/v1/datasetEvents

✅

Get DAG Dataset Queued Event

/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}

✅

Get DAG Dataset Queued Events

/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents

✅

Delete DAG Dataset Queued Event

/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}

✅

Delete DAG Dataset Queued Events

/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents

✅

Get Dataset Queued Events

/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents

✅

Delete Dataset Queued Events

/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents

✅

Monitoring

Get Health

/api/v1/health

✅

DAG Stats

Get DAG Stats

/api/v1/dags/statistics

✅

Config

Get Config

/api/v1/config

✅

Plugins

Get Plugins

/api/v1/plugins

✅

Providers

List Providers

/api/v1/providers

✅

Event Logs

List Event Logs

/api/v1/eventLogs

✅

Get Event Log

/api/v1/eventLogs/{event_log_id}

✅

System

Get Import Errors

/api/v1/importErrors

✅

Get Import Error Details

/api/v1/importErrors/{import_error_id}

✅

Get Health Status

/api/v1/health

✅

Get Version

/api/v1/version

✅

Setup

Dependencies

This project depends on the official Apache Airflow client library (apache-airflow-client). It will be automatically installed when you install this package.

Environment Variables

Set the following environment variables:

AIRFLOW_HOST=<your-airflow-host>        # Optional, defaults to http://localhost:8080
AIRFLOW_TOKEN=<your-airflow-api-token>  # Your Airflow API token
AIRFLOW_API_VERSION=v1                  # Optional, defaults to v1

Basic Authentication (Alternative)

AIRFLOW_HOST=<your-airflow-host>        # Optional, defaults to http://localhost:8080
AIRFLOW_USERNAME=<your-airflow-username>
AIRFLOW_PASSWORD=<your-airflow-password>
AIRFLOW_API_VERSION=v1                  # Optional, defaults to v1

Note: If AIRFLOW_TOKEN is provided, it will be used for authentication. Otherwise, the server will fall back to basic authentication using username and password.

Usage with Claude Desktop

First, clone the repository:

git clone https://github.com/nikhil-ganage/mcp-server-airflow-token

Add to your claude_desktop_config.json:

{
  "mcpServers": {
    "apache-airflow": {
      "type": "stdio",
      "command": "uv",
      "args": [
        "--directory",
        "path-to-repo/mcp-server-airflow-token",
        "run",
        "mcp-server-airflow-token"
      ],
      "env": {
        "AIRFLOW_HOST": "https://astro_id.astronomer.run/id",
        "AIRFLOW_TOKEN": "TOKEN"
      }
    }
  }
}

With Basic Authentication

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

For read-only mode (recommended for safety):

Read-only with Token Authentication

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

Read-only with Basic Authentication

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

Replace path-to-repo with the actual path where you've cloned the repository.

Astronomer Cloud Configuration Example

For Astronomer Cloud deployments:

{
  "mcpServers": {
    "mcp-server-airflow-token": {
      "command": "uvx",
      "args": ["mcp-server-airflow-token"],
      "env": {
        "AIRFLOW_HOST": "https://your-astronomer-domain.astronomer.run/your-deployment-id",
        "AIRFLOW_TOKEN": "your-astronomer-api-token"
      }
    }
  }
}

Note: The deployment ID is part of your Astronomer Cloud URL path.

Selecting the API groups

You can select the API groups you want to use by setting the --apis flag.

uv run mcp-server-airflow-token --apis "dag,dagrun"

The default is to use all APIs.

Allowed values are:

  • config

  • connections

  • dag

  • dagrun

  • dagstats

  • dataset

  • eventlog

  • importerror

  • monitoring

  • plugin

  • pool

  • provider

  • taskinstance

  • variable

  • xcom

Read-Only Mode

You can run the server in read-only mode by using the --read-only flag. This will only expose tools that perform read operations (GET requests) and exclude any tools that create, update, or delete resources.

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

In read-only mode, the server will only expose tools like:

  • Listing DAGs, DAG runs, tasks, variables, connections, etc.

  • Getting details of specific resources

  • Reading configurations and monitoring information

  • Testing connections (non-destructive)

Write operations like creating, updating, deleting DAGs, variables, connections, triggering DAG runs, etc. will not be available in read-only mode.

You can combine read-only mode with API group selection:

uv run mcp-server-airflow-token --read-only --apis "dag,variable"

Manual Execution

You can also run the server manually:

make run

make run accepts following options:

Options:

  • --port: Port to listen on for SSE (default: 8000)

  • --transport: Transport type (stdio/sse, default: stdio)

Or, you could run the sse server directly, which accepts same parameters:

make run-sse

Installation

You can install the server using pip or uvx:

# Using pip
pip install mcp-server-airflow-token

# Using uvx (recommended)
uvx mcp-server-airflow-token

Development

Setting up Development Environment

  1. Clone the repository:

git clone https://github.com/nikhil-ganage/mcp-server-airflow-token.git
cd mcp-server-airflow-token
  1. Install development dependencies:

uv sync --dev
  1. Create a .env file for environment variables (optional for development):

touch .env

Note: No environment variables are required for running tests. The AIRFLOW_HOST defaults to http://localhost:8080 for development and testing purposes.

Running Tests

The project uses pytest for testing with the following commands available:

# Run all tests
make test

Code Quality

# Run linting
make lint

# Run code formatting
make format

Continuous Integration

The project includes a GitHub Actions workflow (.github/workflows/test.yml) that automatically:

  • Runs tests on Python 3.10, 3.11, and 3.12

  • Executes linting checks using ruff

  • Runs on every push and pull request to main branch

The CI pipeline ensures code quality and compatibility across supported Python versions before any changes are merged.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

The package is deployed automatically to PyPI when project.version is updated in pyproject.toml. Follow semver for versioning.

Please include version update in the PR in order to apply the changes to core logic.

License

MIT License

Available Tools

66 tools
clear_dag_runD

Clear a DAG run

Inputs: dag_id (required), dag_run_id (required), dry_run.

clear_task_instancesC

Clear a set of task instances

Inputs: dag_id (required), task_ids, start_date, end_date, +8 more.

create_connectionD

Create a connection

Inputs: conn_id (required), conn_type (required), host, port, +4 more.

create_dataset_eventD

Create dataset event

Inputs: dataset_uri (required), extra.

create_variableD

Create a variable

Inputs: key (required), value (required), description.

delete_connectionC

Delete a connection by ID

Inputs: conn_id (required).

delete_dagC

Delete a DAG

Inputs: dag_id (required).

delete_dag_dataset_queued_eventC

Delete a queued Dataset event for a DAG

Inputs: dag_id (required), uri (required).

delete_dag_dataset_queued_eventsC

Delete queued Dataset events for a DAG

Inputs: dag_id (required), before.

delete_dag_runB

Delete a DAG run by DAG ID and DAG run ID

Inputs: dag_id (required), dag_run_id (required).

delete_dataset_queued_eventsC

Delete queued Dataset events for a Dataset

Inputs: uri (required), before.

delete_poolD

Delete a pool

Inputs: pool_name (required).

delete_variableC

Delete a variable by key

Inputs: key (required).

fetch_dagsC

Fetch all DAGs

Inputs: limit, offset, order_by, tags, +3 more.

get_configD

Get current configuration

Inputs: section.

get_connectionC

Get a connection by ID

Inputs: conn_id (required).

get_dagC

Get a DAG by ID

Inputs: dag_id (required).

get_dag_dataset_queued_eventC

Get a queued Dataset event for a DAG

Inputs: dag_id (required), uri (required).

get_dag_dataset_queued_eventsC

Get queued Dataset events for a DAG

Inputs: dag_id (required).

get_dag_detailsC

Get a simplified representation of DAG

Inputs: dag_id (required), fields.

get_dag_runC

Get a DAG run by DAG ID and DAG run ID

Inputs: dag_id (required), dag_run_id (required).

get_dag_runsD

Get DAG runs by ID

Inputs: dag_id (required), limit, offset, execution_date_gte, +9 more.

get_dag_runs_batchC

List DAG runs (batch)

Inputs: dag_ids, execution_date_gte, execution_date_lte, start_date_gte, +7 more.

get_dag_sourceD

Get a source code

Inputs: file_token (required).

get_dag_statsD

Get DAG stats

Inputs: dag_ids.

get_dag_tasksD

Get tasks for DAG

Inputs: dag_id (required).

get_datasetC

Get a dataset by URI

Inputs: uri (required).

get_dataset_eventsD

Get dataset events

Inputs: limit, offset, order_by, dataset_id, +4 more.

get_dataset_queued_eventsC

Get queued Dataset events for a Dataset

Inputs: uri (required).

get_datasetsD

List datasets

Inputs: limit, offset, order_by, uri_pattern, +1 more.

get_event_logC

Get a specific log entry by ID

Inputs: event_log_id (required).

get_event_logsC

List log entries from event log

Inputs: limit, offset, order_by, dag_id, +10 more.

get_healthC

Get instance status

Inputs: no named parameters.

get_import_errorC

Get a specific import error by ID

Inputs: import_error_id (required).

get_import_errorsC

List import errors

Inputs: limit, offset, order_by.

get_pluginsC

Get a list of loaded plugins

Inputs: limit, offset.

get_poolC

Get a pool by name

Inputs: pool_name (required).

get_poolsD

List pools

Inputs: limit, offset, order_by.

get_providersC

Get a list of loaded providers

Inputs: limit, offset.

get_taskC

Get a task by ID

Inputs: dag_id (required), task_id (required).

get_task_instanceC

Get a task instance by DAG ID, task ID, and DAG run ID

Inputs: dag_id (required), task_id (required), dag_run_id (required).

get_tasksC

Get tasks for DAG

Inputs: dag_id (required), order_by.

get_upstream_dataset_eventsC

Get dataset events for a DAG run

Inputs: dag_id (required), dag_run_id (required).

get_valueC

Get a specific option from configuration

Inputs: section (required), option (required).

get_variableC

Get a variable by key

Inputs: key (required).

get_versionC

Get version information

Inputs: no named parameters.

get_xcom_entriesD

Get all XCom entries

Inputs: dag_id (required), dag_run_id (required), task_id (required), map_index, +3 more.

get_xcom_entryD

Get an XCom entry

Inputs: dag_id (required), dag_run_id (required), task_id (required), xcom_key (required), +3 more.

list_connectionsC

List all connections

Inputs: limit, offset, order_by.

list_task_instancesC

List task instances by DAG ID and DAG run ID

Inputs: dag_id (required), dag_run_id (required), execution_date_gte, execution_date_lte, +13 more.

list_variablesD

List all variables

Inputs: limit, offset, order_by.

patch_dagD

Update a DAG

Inputs: dag_id (required), is_paused, tags.

patch_dagsC

Update multiple DAGs

Inputs: dag_id_pattern, is_paused, tags.

patch_poolC

Update a pool

Inputs: pool_name (required), slots, description, include_deferred.

pause_dagC

Pause a DAG by ID

Inputs: dag_id (required).

post_dag_runC

Trigger a DAG by ID

Inputs: dag_id (required), dag_run_id, data_interval_end, data_interval_start, +8 more.

post_poolD

Create a pool

Inputs: name (required), slots (required), description, include_deferred.

reparse_dag_fileC

Request re-parsing of a DAG file

Inputs: file_token (required).

set_dag_run_noteC

Update the DagRun note

Inputs: dag_id (required), dag_run_id (required), note (required).

set_task_instances_stateD

Set a state of task instances

Inputs: dag_id (required), state (required), task_ids, execution_date, +5 more.

test_connectionD

Test a connection

Inputs: conn_type (required), host, port, login, +3 more.

unpause_dagC

Unpause a DAG by ID

Inputs: dag_id (required).

update_connectionC

Update a connection by ID

Inputs: conn_id (required), conn_type, host, port, +4 more.

update_dag_run_stateC

Update a DAG run state by DAG ID and DAG run ID

Inputs: dag_id (required), dag_run_id (required), state.

update_task_instanceC

Update a task instance by DAG ID, DAG run ID, and task ID

Inputs: dag_id (required), dag_run_id (required), task_id (required), state.

update_variableC

Update a variable by key

Inputs: key (required), value, description.

Tool Schema Changelog

Recent tool additions, removals, and schema changes observed during successful MCP inspections.

Loading tool schema changelog…

    TDQS

    C2.5/5.0

    Scored across 66 tools

    Disambiguation4/5

    Most tools have distinct purposes targeting specific resources and actions, such as clear_dag_run vs. delete_dag_run. However, some overlap exists, like get_dag_details and get_dag, which could cause confusion about which to use for DAG information, and get_tasks and get_dag_tasks appear redundant.

    Naming Consistency5/5

    Tool names follow a highly consistent verb_noun pattern throughout, using snake_case uniformly. Examples include create_connection, delete_dag, get_config, and update_variable, making them predictable and easy to parse.

    Tool Count2/5

    With 66 tools, the count is excessive for typical MCP server use, likely overwhelming for agents and indicating poor scoping. While Airflow is complex, this many tools suggests fragmentation rather than a streamlined interface.

    Completeness5/5

    The tool set provides comprehensive CRUD and lifecycle coverage for Airflow's domain, including DAGs, tasks, connections, variables, datasets, and more. No obvious gaps are present; it supports full management workflows from creation to deletion and monitoring.

    Maintenance

    ActivityInactive
    ResponsivenessNo recent community issues

    Related MCP Connectors

    Related MCP Servers

    • A
      license
      B
      quality
      F
      maintenance
      A Model Context Protocol server that allows AI assistants to interact with Prefect's workflow automation platform through natural language, enabling users to manage flows, deployments, tasks, and other Prefect resources via conversational commands.
      46
      15
      Apache 2.0
    • -
      license
      Not graded
      quality
      Not graded
      maintenance
      A Model Context Protocol server that provides access to messages from an external API endpoint using Bearer token authentication for integration with Claude Desktop.
      -
    • A
      license
      A
      quality
      D
      maintenance
      A production-ready Model Context Protocol server that enables AI assistants and applications to interact with Airtable bases through a standardized interface with secure OAuth 2.0 authentication.
      10
      3
      MIT
    • A
      license
      Not graded
      quality
      D
      maintenance
      Enables interaction with Apache Airflow through the Model Context Protocol, allowing users to manage DAGs, task instances, variables, connections, pools, XComs, and datasets.
      MIT