style: ruff lint and format

Author: yangkyeongmo

.gitignore

@@ -20,6 +20,7 @@ wheels/
 .installed.cfg
 *.egg
 .pypirc
+.ruff_cache/
 
 # Virtual Environment
 .env

.tool-versions

@@ -1 +1 @@
-python 3.12.9
+python 3.12.6

Makefile

@@ -1,5 +1,8 @@
 build:
 	uv run python -m build
-
 publish:
 	uv run python -m twine upload --config-file .pypirc dist/*
+lint:
+	uv run ruff check . --fix
+format:
+	uv run ruff format .

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mcp-server-apache-airflow"
-version = "0.1.0a1"
+version = "0.1.3"
 description = "Model Context Protocol (MCP) server for Apache Airflow"
 authors = [
     { name = "Gyeongmo Yang", email = "[email protected]" }
@@ -20,6 +20,7 @@ classifiers = [
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
     "Topic :: Software Development :: Libraries :: Python Modules",
 ]
 keywords = ["mcp", "airflow", "apache-airflow", "model-context-protocol"]
@@ -51,3 +52,7 @@ include = [
     "README.md",
     "LICENSE",
 ]
+
+[tool.ruff]
+select = ["E", "W", "F", "B", "I"]
+line-length = 120

src/airflow.py

@@ -1,21 +1,21 @@
-import os
-
 import httpx
 import mcp.types as types
 
-AIRFLOW_HOST = os.getenv("AIRFLOW_HOST").rstrip("/")
-AIRFLOW_USERNAME = os.getenv("AIRFLOW_USERNAME")
-AIRFLOW_PASSWORD = os.getenv("AIRFLOW_PASSWORD")
+from src.envs import AIRFLOW_HOST, AIRFLOW_PASSWORD, AIRFLOW_USERNAME
+
 
 def get_dag_url(dag_id: str) -> str:
     return f"{AIRFLOW_HOST}/dags/{dag_id}/grid"
 
+
 def get_dag_run_url(dag_id: str, dag_run_id: str) -> str:
     return f"{AIRFLOW_HOST}/dags/{dag_id}/grid?dag_run_id={dag_run_id}"
 
+
 def get_task_instance_url(dag_id: str, dag_run_id: str, task_id: str) -> str:
     return f"{AIRFLOW_HOST}/dags/{dag_id}/grid?dag_run_id={dag_run_id}&task_id={task_id}"
 
+
 async def fetch_dags(
     limit: int | None = None,
     offset: int | None = None,
@@ -44,69 +44,62 @@ async def fetch_dags(
 
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
-        response = await client.get(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            params=params
-        )
+        response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), params=params)
         response.raise_for_status()
         data = response.json()
-        
+
         # Add UI links to each DAG
         for dag in data.get("dags", []):
             dag["ui_url"] = get_dag_url(dag["dag_id"])
-            
+
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def get_dag(dag_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD))
         response.raise_for_status()
         data = response.json()
-        
+
         # Add UI link to DAG
         data["ui_url"] = get_dag_url(dag_id)
-        
+
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def pause_dag(dag_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.patch(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            json={"is_paused": True}
+            url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), json={"is_paused": True}
         )
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def unpause_dag(dag_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.patch(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            json={"is_paused": False}
+            url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), json={"is_paused": False}
         )
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def trigger_dag(dag_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}/dagRuns"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
-        response = await client.post(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            json={}
-        )
+        response = await client.post(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), json={})
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def get_dag_runs(
     dag_id: str,
     limit: int | None = None,
@@ -151,20 +144,17 @@ async def get_dag_runs(
 
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
-        response = await client.get(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            params=params
-        )
+        response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), params=params)
         response.raise_for_status()
         data = response.json()
-        
+
         # Add UI links to each DAG run
         for dag_run in data.get("dag_runs", []):
             dag_run["ui_url"] = get_dag_run_url(dag_id, dag_run["dag_run_id"])
-            
+
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def get_dag_tasks(dag_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}/tasks"
     url = f"{AIRFLOW_HOST}{path}"
@@ -173,14 +163,18 @@ async def get_dag_tasks(dag_id: str) -> list[types.TextContent | types.ImageCont
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
-async def get_task_instance(dag_id: str, task_id: str, dag_run_id: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
+
+async def get_task_instance(
+    dag_id: str, task_id: str, dag_run_id: str
+) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD))
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def list_task_instances(
     dag_id: str,
     dag_run_id: str,
@@ -235,22 +229,22 @@ async def list_task_instances(
 
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
-        response = await client.get(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            params=params
-        )
+        response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), params=params)
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
-async def get_import_error(import_error_id: int) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
+
+async def get_import_error(
+    import_error_id: int,
+) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = f"/api/v1/importErrors/{import_error_id}"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD))
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def list_import_errors(
     limit: int | None = None,
     offset: int | None = None,
@@ -267,14 +261,11 @@ async def list_import_errors(
 
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
-        response = await client.get(
-            url, 
-            auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD),
-            params=params
-        )
+        response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD), params=params)
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def get_health() -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = "/api/v1/health"
     url = f"{AIRFLOW_HOST}{path}"
@@ -283,10 +274,11 @@ async def get_health() -> list[types.TextContent | types.ImageContent | types.Em
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
 
+
 async def get_version() -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
     path = "/api/v1/version"
     url = f"{AIRFLOW_HOST}{path}"
     async with httpx.AsyncClient(follow_redirects=True) as client:
         response = await client.get(url, auth=httpx.BasicAuth(AIRFLOW_USERNAME, AIRFLOW_PASSWORD))
         response.raise_for_status()
-        return [types.TextContent(type="text", text=response.text)] 
+        return [types.TextContent(type="text", text=response.text)]

src/envs.py

@@ -0,0 +1,5 @@
+import os
+
+AIRFLOW_HOST = os.getenv("AIRFLOW_HOST").rstrip("/")
+AIRFLOW_USERNAME = os.getenv("AIRFLOW_USERNAME")
+AIRFLOW_PASSWORD = os.getenv("AIRFLOW_PASSWORD")