get_import_errors
Retrieve and list import errors from Apache Airflow deployments to identify and troubleshoot DAG loading issues, with options to limit, offset, and order results.
Instructions
List import errors
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| offset | No | ||
| order_by | No |
Input Schema (JSON Schema)
{
"properties": {
"limit": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"title": "Limit"
},
"offset": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"title": "Offset"
},
"order_by": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Order By"
}
},
"type": "object"
}
Implementation Reference
- src/airflow/importerror.py:19-35 (handler)The async handler function that executes the tool logic: accepts optional parameters limit, offset, order_by; calls Airflow's ImportErrorApi.get_import_errors; returns the response as a TextContent list.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)Module-level registration function that provides the get_import_errors tool details (function reference, name, description, read-only flag) for top-level registration.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/main.py:12-12 (registration)Import of the module's get_all_functions for inclusion in the central tool registry.from src.airflow.importerror import get_all_functions as get_importerror_functions
- src/airflow/importerror.py:3-8 (helper)Initialization of the ImportErrorApi client instance used by the handler.import mcp.types as types from airflow_client.client.api.import_error_api import ImportErrorApi from src.airflow.airflow_client import api_client import_error_api = ImportErrorApi(api_client)
- src/main.py:90-92 (registration)The generic tool addition loop in main() that registers all tools, including get_import_errors, to the FastMCP app.for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description)