MCP Server for Apache Airflow

Instructions

Implementation Reference

• src/airflow/xcom.py:43-64 (handler)

  The main handler function that executes the tool's logic: builds kwargs from optional params and calls the underlying XComApi to retrieve and return the XCom entry as text.

  async def get_xcom_entry(
      dag_id: str,
      dag_run_id: str,
      task_id: str,
      xcom_key: str,
      map_index: Optional[int] = None,
      deserialize: Optional[bool] = None,
      stringify: Optional[bool] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      # Build parameters dictionary
      kwargs: Dict[str, Any] = {}
      if map_index is not None:
          kwargs["map_index"] = map_index
      if deserialize is not None:
          kwargs["deserialize"] = deserialize
      if stringify is not None:
          kwargs["stringify"] = stringify
  
      response = xcom_api.get_xcom_entry(
          dag_id=dag_id, dag_run_id=dag_run_id, task_id=task_id, xcom_key=xcom_key, **kwargs
      )
      return [types.TextContent(type="text", text=str(response.to_dict()))]

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

  Helper function that provides the registration tuple for the get_xcom_entry tool, used by main.py to add it to the MCP app.

  def get_all_functions() -> list[tuple[Callable, str, str, bool]]:
      """Return list of (function, name, description, is_read_only) tuples for registration."""
      return [
          (get_xcom_entries, "get_xcom_entries", "Get all XCom entries", True),
          (get_xcom_entry, "get_xcom_entry", "Get an XCom entry", True),
      ]

• src/main.py:20-20 (registration)

  Imports the get_all_functions from xcom module to enable registration of xcom tools including get_xcom_entry.

  from src.airflow.xcom import get_all_functions as get_xcom_functions

• src/main.py:24-40 (registration)

  Maps APIType.XCOM to the xcom get_all_functions, which includes get_xcom_entry, for use in registering tools.

  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,
  }

• src/main.py:95-97 (registration)

  Generic loop that adds each tool from the functions list (including get_xcom_entry when XCOM API is selected) to the MCP app using Tool.from_function.

  for func, name, description, *_ in functions:
      app.add_tool(Tool.from_function(func, name=name, description=description))