MCP Server Airflow Token

Instructions

Implementation Reference

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

  The async handler function that implements the logic for the 'get_xcom_entry' tool by calling the Airflow XCom API and formatting the response as MCP TextContent.

  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)

  The registration function that provides the tuple for the 'get_xcom_entry' tool, including its name, description, and read-only flag, which is imported and used in main.py to register the tool via app.add_tool.

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