MCP Server for Apache Airflow

Instructions

Implementation Reference

• src/airflow/dagstats.py:18-27 (handler)

  The asynchronous handler function that executes the 'get_dag_stats' tool. It accepts an optional list of DAG IDs, calls the underlying Airflow DagStatsApi, and returns the stats as a text content response.
  async def get_dag_stats( dag_ids: Optional[List[str]] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: # Build parameters dictionary kwargs: Dict[str, Any] = {} if dag_ids is not None: kwargs["dag_ids"] = dag_ids response = dag_stats_api.get_dag_stats(**kwargs) return [types.TextContent(type="text", text=str(response.to_dict()))]
• src/airflow/dagstats.py:11-16 (registration)

  Provides the registration tuple for the 'get_dag_stats' tool, which is used by the central registration in main.py to add the tool via app.add_tool.
  def get_all_functions() -> list[tuple[Callable, str, str, bool]]: """Return list of (function, name, description, is_read_only) tuples for registration.""" return [ (get_dag_stats, "get_dag_stats", "Get DAG stats", True), ]
• src/main.py:95-96 (registration)

  The generic loop in the main entrypoint that registers all tools, including 'get_dag_stats', by calling Tool.from_function and adding to the MCP app.
  for func, name, description, *_ in functions: app.add_tool(Tool.from_function(func, name=name, description=description))
• src/airflow/dagstats.py:3-8 (helper)

  Imports and initializes the DagStatsApi instance using the shared api_client, which the handler uses to fetch DAG stats.
  import mcp.types as types from airflow_client.client.api.dag_stats_api import DagStatsApi from src.airflow.airflow_client import api_client dag_stats_api = DagStatsApi(api_client)