MCP Server Airflow Token

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/airflow/connection.py:23-39 (handler)

  The main handler function for the 'list_connections' tool. It accepts optional parameters limit, offset, and order_by, constructs a kwargs dict, calls connection_api.get_connections, and returns the response as a TextContent object.
  async def list_connections( 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 = connection_api.get_connections(**kwargs) return [types.TextContent(type="text", text=str(response.to_dict()))]
• src/airflow/connection.py:11-20 (registration)

  Local registration of the list_connections tool (and others) via get_all_functions, which returns a tuple including the function reference, name, description, and 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 [ (list_connections, "list_connections", "List all connections", True), (create_connection, "create_connection", "Create a connection", False), (get_connection, "get_connection", "Get a connection by ID", True), (update_connection, "update_connection", "Update a connection by ID", False), (delete_connection, "delete_connection", "Delete a connection by ID", False), (test_connection, "test_connection", "Test a connection", True), ]
• src/main.py:6-98 (registration)

  Global registration in main.py: imports get_all_functions from connection.py, maps it in APITYPE_TO_FUNCTIONS, calls it to get the list of functions, and registers each with app.add_tool using the name and description.
  from src.airflow.connection import get_all_functions as get_connection_functions from src.airflow.dag import get_all_functions as get_dag_functions from src.airflow.dagrun import get_all_functions as get_dagrun_functions from src.airflow.dagstats import get_all_functions as get_dagstats_functions from src.airflow.dataset import get_all_functions as get_dataset_functions from src.airflow.eventlog import get_all_functions as get_eventlog_functions from src.airflow.importerror import get_all_functions as get_importerror_functions from src.airflow.monitoring import get_all_functions as get_monitoring_functions from src.airflow.plugin import get_all_functions as get_plugin_functions from src.airflow.pool import get_all_functions as get_pool_functions from src.airflow.provider import get_all_functions as get_provider_functions from src.airflow.taskinstance import get_all_functions as get_taskinstance_functions from src.airflow.variable import get_all_functions as get_variable_functions from src.airflow.xcom import get_all_functions as get_xcom_functions from src.enums import APIType APITYPE_TO_FUNCTIONS = { APIType.CONFIG: get_config_functions, APIType.CONNECTION: get_connection_functions, APIType.DAG: get_dag_functions, APIType.DAGRUN: get_dagrun_functions, APIType.DAGSTATS: get_dagstats_functions, APIType.DATASET: get_dataset_functions, APIType.EVENTLOG: get_eventlog_functions, APIType.IMPORTERROR: get_importerror_functions, APIType.MONITORING: get_monitoring_functions, APIType.PLUGIN: get_plugin_functions, APIType.POOL: get_pool_functions, APIType.PROVIDER: get_provider_functions, APIType.TASKINSTANCE: get_taskinstance_functions, APIType.VARIABLE: get_variable_functions, APIType.XCOM: get_xcom_functions, } def filter_functions_for_read_only(functions: list[tuple]) -> list[tuple]: """ Filter functions to only include read-only operations. Args: functions: List of (func, name, description, is_read_only) tuples Returns: List of (func, name, description, is_read_only) tuples with only read-only functions """ return [ (func, name, description, is_read_only) for func, name, description, is_read_only in functions if is_read_only ] @click.command() @click.option( "--transport", type=click.Choice(["stdio", "sse"]), default="stdio", help="Transport type", ) @click.option( "--apis", type=click.Choice([api.value for api in APIType]), default=[api.value for api in APIType], multiple=True, help="APIs to run, default is all", ) @click.option( "--read-only", is_flag=True, help="Only expose read-only tools (GET operations, no CREATE/UPDATE/DELETE)", ) def main(transport: str, apis: list[str], read_only: bool) -> None: from src.server import app for api in apis: logging.debug(f"Adding API: {api}") get_function = APITYPE_TO_FUNCTIONS[APIType(api)] try: functions = get_function() except NotImplementedError: continue # Filter functions for read-only mode if requested if read_only: functions = filter_functions_for_read_only(functions) for func, name, description, *_ in functions: app.add_tool(func, name=name, description=description) if transport == "sse": logging.debug("Starting MCP server for Apache Airflow with SSE transport") app.run(transport="sse") else: logging.debug("Starting MCP server for Apache Airflow with stdio transport") app.run(transport="stdio")