MCP Server for Apache Airflow

Instructions

Implementation Reference

• src/airflow/importerror.py:19-34 (handler)

  The async handler function that implements the get_import_errors tool. It accepts optional parameters limit, offset, order_by, calls the Airflow ImportErrorApi, and formats the response as MCP TextContent.

  async def get_import_errors(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      order_by: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      # Build parameters dictionary
      kwargs: Dict[str, Any] = {}
      if limit is not None:
          kwargs["limit"] = limit
      if offset is not None:
          kwargs["offset"] = offset
      if order_by is not None:
          kwargs["order_by"] = order_by
  
      response = import_error_api.get_import_errors(**kwargs)
      return [types.TextContent(type="text", text=str(response.to_dict()))]

• src/airflow/importerror.py:11-16 (registration)

  Function providing the registration tuple for the get_import_errors MCP tool (function, name, description, read-only flag).

  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),
      ]