get_health
Check the operational status of an Apache Airflow deployment to verify system health and availability for workflow management.
Instructions
Get instance status
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/airflow/monitoring.py:19-25 (handler)The async handler function that implements the 'get_health' tool logic. It calls the Airflow monitoring API to fetch health status and returns it formatted as MCP TextContent.
async def get_health() -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get the status of Airflow's metadatabase, triggerer and scheduler. It includes info about metadatabase and last heartbeat of scheduler and triggerer. """ response = monitoring_api.get_health() return [types.TextContent(type="text", text=str(response.to_dict()))] - src/airflow/monitoring.py:11-16 (registration)The get_all_functions() function that registers the 'get_health' tool by providing its handler function, name, description, and read-only status, which is later used 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_health, "get_health", "Get instance status", True), (get_version, "get_version", "Get version information", True), ] - src/airflow/monitoring.py:8-8 (helper)Initialization of the monitoring_api client instance used by the get_health handler to query Airflow's health endpoint.
monitoring_api = MonitoringApi(api_client) - src/main.py:90-92 (registration)The generic tool registration loop in main.py where functions from get_all_functions() (including get_health) are added to the MCP app using app.add_tool().
for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)