MCP Server Airflow Token

Instructions

Implementation Reference

• src/airflow/importerror.py:37-41 (handler)

  The async handler function that implements the core logic of the 'get_import_error' tool. It takes an import_error_id, calls the Airflow ImportErrorApi to retrieve the specific import error, and returns it formatted as MCP TextContent.
  async def get_import_error( import_error_id: int, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: response = import_error_api.get_import_error(import_error_id=import_error_id) return [types.TextContent(type="text", text=str(response.to_dict()))]
• src/airflow/importerror.py:11-16 (registration)

  The module-level get_all_functions that returns the registration tuple for 'get_import_error', linking the handler function to its name and description for inclusion in the central MCP tool registry.
  def get_all_functions() -> list[tuple[Callable, str, str, bool]]: """Return list of (function, name, description, is_read_only) tuples for registration.""" return [ (get_import_errors, "get_import_errors", "List import errors", True), (get_import_error, "get_import_error", "Get a specific import error by ID", True), ]
• src/airflow/importerror.py:8-8 (helper)

  Initialization of the ImportErrorApi client instance used by the get_import_error handler.
  import_error_api = ImportErrorApi(api_client)
• src/main.py:90-92 (registration)

  The generic registration loop in main.py where all tools, including get_import_error, are added to the MCP app via app.add_tool using data from module get_all_functions.
  for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)
• src/main.py:12-12 (registration)

  Import of the get_all_functions from importerror.py into the central main.py registry.
  from src.airflow.importerror import get_all_functions as get_importerror_functions