MCP-Airflow-API

Instructions

Implementation Reference

• src/mcp_airflow_api/tools/common_tools.py:263-278 (handler)

  Core handler function for the 'trigger_dag' tool. Decorated with @mcp.tool() for automatic registration. Triggers a new DAG run by making a POST request to the Airflow API.

  @mcp.tool()
  async def trigger_dag(dag_id: str) -> Dict[str, Any]:
      """[Tool Role]: Triggers a new DAG run for a specified Airflow DAG."""
      if not dag_id:
          raise ValueError("dag_id must not be empty")
      resp = await airflow_request("POST", f"/dags/{dag_id}/dagRuns", json={"conf": {}})
      resp.raise_for_status()
      run = resp.json()
      return {
          "dag_id": dag_id,
          "run_id": run.get("run_id"),
          "state": run.get("state"),
          "execution_date": run.get("execution_date"),
          "start_date": run.get("start_date"),
          "end_date": run.get("end_date")
      }

• src/mcp_airflow_api/tools/v1_tools.py:20-23 (registration)

  Registration of common tools (including trigger_dag) for Airflow API v1 by calling register_common_tools after setting v1-specific airflow_request.

  common_tools.airflow_request = airflow_request_v1
  
  # Register all 56 common tools (includes management tools)
  common_tools.register_common_tools(mcp)

• src/mcp_airflow_api/tools/v2_tools.py:21-24 (registration)

  Registration of common tools (including trigger_dag) for Airflow API v2 by calling register_common_tools after setting v2-specific airflow_request.

  common_tools.airflow_request = airflow_request_v2
  
  # Register all 43 common tools
  common_tools.register_common_tools(mcp)

• src/mcp_airflow_api/mcp_main.py:265-266 (registration)

  Top-level registration call for v1 tools in mcp_main.py (create_mcp_server function).

  from mcp_airflow_api.tools import v1_tools
  v1_tools.register_tools(mcp_instance)

• src/mcp_airflow_api/mcp_main.py:269-270 (registration)

  Top-level registration call for v2 tools in mcp_main.py (create_mcp_server function).

  from mcp_airflow_api.tools import v2_tools
  v2_tools.register_tools(mcp_instance)