get_import_error
Retrieve specific import error details from Apache Airflow deployments by providing the error ID to diagnose and resolve data pipeline issues.
Instructions
Get a specific import error by ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| import_error_id | Yes |
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