get_dag_stats
Retrieve statistics for Apache Airflow DAGs to monitor performance and status through the MCP Server for Apache Airflow.
Instructions
Get DAG stats
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| dag_ids | No |
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)