🚀 mcp-server-airflow-token
這是一個支持Bearer令牌認證的Apache Airflow的模型上下文協議(MCP)服務器,可實現與Astronomer Cloud和獨立Airflow實例的無縫集成。
基於Gyeongmo Nathan Yang的mcp-server-apache-airflow
此分支通過添加Bearer令牌認證支持對原始MCP服務器進行了增強,使其與Astronomer Cloud和其他基於令牌的Airflow部署兼容。
✨ 主要特性
- ✅ Bearer令牌認證 - 現代Airflow部署的主要認證方法
- ✅ 與Astronomer Cloud兼容 - 可與Astronomer的託管Airflow無縫協作
- ✅ 向後兼容 - 仍然支持用戶名/密碼認證
- ✅ 增強的URL處理 - 正確處理諸如
/deployment-id
之類的部署路徑
📚 詳細文檔
本項目實現了一個模型上下文協議服務器,該服務器封裝了Apache Airflow的REST API,允許MCP客戶端以標準化方式與Airflow進行交互。它使用官方的Apache Airflow客戶端庫來確保兼容性和可維護性。
特性實現狀態
特性 |
API路徑 |
狀態 |
DAG管理 |
|
|
列出DAG |
/api/v1/dags |
✅ |
獲取DAG詳情 |
/api/v1/dags/{dag_id} |
✅ |
暫停DAG |
/api/v1/dags/{dag_id} |
✅ |
恢復DAG |
/api/v1/dags/{dag_id} |
✅ |
更新DAG |
/api/v1/dags/{dag_id} |
✅ |
刪除DAG |
/api/v1/dags/{dag_id} |
✅ |
獲取DAG源 |
/api/v1/dagSources/{file_token} |
✅ |
批量更新DAG |
/api/v1/dags |
✅ |
重新解析DAG文件 |
/api/v1/dagSources/{file_token}/reparse |
✅ |
DAG運行 |
|
|
列出DAG運行 |
/api/v1/dags/{dag_id}/dagRuns |
✅ |
創建DAG運行 |
/api/v1/dags/{dag_id}/dagRuns |
✅ |
獲取DAG運行詳情 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id} |
✅ |
更新DAG運行 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id} |
✅ |
刪除DAG運行 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id} |
✅ |
批量獲取DAG運行 |
/api/v1/dags/~/dagRuns/list |
✅ |
清除DAG運行 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clear |
✅ |
設置DAG運行備註 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/setNote |
✅ |
獲取上游數據集事件 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/upstreamDatasetEvents |
✅ |
任務 |
|
|
列出DAG任務 |
/api/v1/dags/{dag_id}/tasks |
✅ |
獲取任務詳情 |
/api/v1/dags/{dag_id}/tasks/{task_id} |
✅ |
獲取任務實例 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id} |
✅ |
列出任務實例 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances |
✅ |
更新任務實例 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id} |
✅ |
清除任務實例 |
/api/v1/dags/{dag_id}/clearTaskInstances |
✅ |
設置任務實例狀態 |
/api/v1/dags/{dag_id}/updateTaskInstancesState |
✅ |
變量 |
|
|
列出變量 |
/api/v1/variables |
✅ |
創建變量 |
/api/v1/variables |
✅ |
獲取變量 |
/api/v1/variables/{variable_key} |
✅ |
更新變量 |
/api/v1/variables/{variable_key} |
✅ |
刪除變量 |
/api/v1/variables/{variable_key} |
✅ |
連接 |
|
|
列出連接 |
/api/v1/connections |
✅ |
創建連接 |
/api/v1/connections |
✅ |
獲取連接 |
/api/v1/connections/{connection_id} |
✅ |
更新連接 |
/api/v1/connections/{connection_id} |
✅ |
刪除連接 |
/api/v1/connections/{connection_id} |
✅ |
測試連接 |
/api/v1/connections/test |
✅ |
池 |
|
|
列出池 |
/api/v1/pools |
✅ |
創建池 |
/api/v1/pools |
✅ |
獲取池 |
/api/v1/pools/{pool_name} |
✅ |
更新池 |
/api/v1/pools/{pool_name} |
✅ |
刪除池 |
/api/v1/pools/{pool_name} |
✅ |
XComs |
|
|
列出XComs |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries |
✅ |
獲取XCom條目 |
/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key} |
✅ |
數據集 |
|
|
列出數據集 |
/api/v1/datasets |
✅ |
獲取數據集 |
/api/v1/datasets/{uri} |
✅ |
獲取數據集事件 |
/api/v1/datasetEvents |
✅ |
創建數據集事件 |
/api/v1/datasetEvents |
✅ |
獲取DAG數據集排隊事件 |
/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri} |
✅ |
獲取DAG數據集排隊事件列表 |
/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents |
✅ |
刪除DAG數據集排隊事件 |
/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri} |
✅ |
刪除DAG數據集排隊事件列表 |
/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents |
✅ |
獲取數據集排隊事件 |
/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents |
✅ |
刪除數據集排隊事件 |
/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents |
✅ |
監控 |
|
|
獲取健康狀態 |
/api/v1/health |
✅ |
DAG統計 |
|
|
獲取DAG統計信息 |
/api/v1/dags/statistics |
✅ |
配置 |
|
|
獲取配置 |
/api/v1/config |
✅ |
插件 |
|
|
獲取插件 |
/api/v1/plugins |
✅ |
提供者 |
|
|
列出提供者 |
/api/v1/providers |
✅ |
事件日誌 |
|
|
列出事件日誌 |
/api/v1/eventLogs |
✅ |
獲取事件日誌 |
/api/v1/eventLogs/{event_log_id} |
✅ |
系統 |
|
|
獲取導入錯誤 |
/api/v1/importErrors |
✅ |
獲取導入錯誤詳情 |
/api/v1/importErrors/{import_error_id} |
✅ |
獲取健康狀態 |
/api/v1/health |
✅ |
獲取版本信息 |
/api/v1/version |
✅ |
📦 安裝指南
依賴項
本項目依賴於官方的Apache Airflow客戶端庫(apache-airflow-client
)。在安裝此軟件包時,它將自動安裝。
環境變量
設置以下環境變量:
令牌認證(推薦)
AIRFLOW_HOST=<your-airflow-host> # 可選,默認為 http://localhost:8080
AIRFLOW_TOKEN=<your-airflow-api-token> # 你的Airflow API令牌
AIRFLOW_API_VERSION=v1 # 可選,默認為 v1
基本認證(可選)
AIRFLOW_HOST=<your-airflow-host> # 可選,默認為 http://localhost:8080
AIRFLOW_USERNAME=<your-airflow-username>
AIRFLOW_PASSWORD=<your-airflow-password>
AIRFLOW_API_VERSION=v1 # 可選,默認為 v1
注意:如果提供了AIRFLOW_TOKEN
,將使用它進行認證。否則,服務器將回退到使用用戶名和密碼的基本認證。
與Claude Desktop配合使用
首先,克隆倉庫:
git clone https://github.com/nikhil-ganage/mcp-server-airflow-token
將以下內容添加到你的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"
}
}
}
}
使用基本認證
{
"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"
}
}
}
}
對於只讀模式(出於安全考慮推薦):
只讀模式下使用令牌認證
{
"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"
}
}
}
}
只讀模式下使用基本認證
{
"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"
}
}
}
}
將path-to-repo
替換為你克隆倉庫的實際路徑。
Astronomer Cloud配置示例
對於Astronomer Cloud部署:
{
"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"
}
}
}
}
注意:部署ID是Astronomer Cloud URL路徑的一部分。
選擇API組
你可以通過設置--apis
標誌來選擇要使用的API組。
uv run mcp-server-airflow-token --apis "dag,dagrun"
默認情況下使用所有API。
允許的值包括:
- config
- connections
- dag
- dagrun
- dagstats
- dataset
- eventlog
- importerror
- monitoring
- plugin
- pool
- provider
- taskinstance
- variable
- xcom
只讀模式
你可以使用--read-only
標誌以只讀模式運行服務器。這將僅暴露執行讀取操作(GET請求)的工具,並排除任何創建、更新或刪除資源的工具。
uv run mcp-server-airflow-token --read-only
在只讀模式下,服務器將僅暴露以下工具:
- 列出DAG、DAG運行、任務、變量、連接等
- 獲取特定資源的詳細信息
- 讀取配置和監控信息
- 測試連接(非破壞性)
諸如創建、更新、刪除DAG、變量、連接、觸發DAG運行等寫操作在只讀模式下將不可用。
你可以將只讀模式與API組選擇結合使用:
uv run mcp-server-airflow-token --read-only --apis "dag,variable"
手動執行
你也可以手動運行服務器:
make run
make run
接受以下選項:
--port
:SSE監聽端口(默認:8000)
--transport
:傳輸類型(stdio/sse,默認:stdio)
或者,你可以直接運行sse服務器,它接受相同的參數:
make run-sse
安裝
你可以使用pip或uvx安裝服務器:
pip install mcp-server-airflow-token
uvx mcp-server-airflow-token
🔧 開發
設置開發環境
- 克隆倉庫:
git clone https://github.com/nikhil-ganage/mcp-server-airflow-token.git
cd mcp-server-airflow-token
- 安裝開發依賴項:
uv sync --dev
- 創建一個
.env
文件用於設置環境變量(開發時可選):
touch .env
注意:運行測試不需要設置環境變量。出於開發和測試目的,AIRFLOW_HOST
默認為http://localhost:8080
。
運行測試
項目使用pytest進行測試,可使用以下命令:
make test
代碼質量
make lint
make format
持續集成
項目包含一個GitHub Actions工作流(.github/workflows/test.yml
),它會自動執行以下操作:
- 在Python 3.10、3.11和3.12上運行測試
- 使用ruff執行代碼檢查
- 在每次推送到
main
分支和拉取請求時運行
CI管道可確保在合併任何更改之前,代碼質量和在支持的Python版本上的兼容性。
貢獻
歡迎貢獻代碼!請隨時提交拉取請求。
當pyproject.toml
中的project.version
更新時,該軟件包將自動部署到PyPI。請遵循語義化版本控制。
請在拉取請求中包含版本更新,以便將更改應用到核心邏輯。
📄 許可證
MIT許可證