TFX TypeError: Argument input_params should be a Channel of type <class 'tfx.types.standard_artifacts.ExternalArtifact'> (got test_string)

Question

I updated the TFX from 0.22.0 to 1.8.0 and now I'm getting this error:

TypeError: Argument input_params should be a Channel of type <class 'tfx.types.standard_artifacts.ExternalArtifact'> (got test_string)

With 0.22.0 I used external_input function like this and worked:

from tfx.utils.dsl_utils import external_input
my_componet = MyCustomComponent(input_params=external_input("test_string"))

But after upgrade tfx not work anymore. My code is:

import os
import absl
from typing import Any, Dict, List, Optional, Text
from tfx.components.base.base_component import BaseComponent
from tfx.components.base.executor_spec import ExecutorClassSpec
from tfx.components.example_gen import utils
from tfx.types import Channel, channel_utils, standard_artifacts, Artifact
from tfx.types.component_spec import ComponentSpec, ExecutionParameter, ChannelParameter
from tfx.proto import example_gen_pb2
from tfx.components.base.base_executor import BaseExecutor
from tfx.orchestration import metadata, pipeline
from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner

_pipeline_name = 'debug_custom_pipeline'
_taxi_root = os.path.join(os.environ['HOME'], 'debug')
_data_root = os.path.join(os.path.dirname(__file__), 'data', 'simple')
_tfx_root = os.path.join(_taxi_root, 'tfx')
_pipeline_root = os.path.join(_tfx_root, 'pipelines', _pipeline_name)
_metadata_path = os.path.join(_tfx_root, 'metadata', _pipeline_name, 'metadata.db')


def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     metadata_path: str) -> pipeline.Pipeline:
  print("creating pipeline")
  my_componet = MyCustomComponent(input_params="test_string")

  return pipeline.Pipeline(
      pipeline_name=pipeline_name,
      pipeline_root=pipeline_root,
      components=[my_componet],
      enable_cache=True,
      metadata_connection_config=metadata.sqlite_metadata_connection_config(metadata_path))


class MyCustomExecutor(BaseExecutor):
    """Executor for ModelExporter component."""

    def Do(self, input_dict: Dict[Text, List[Artifact]],
         output_dict: Dict[Text, List[Artifact]],
         exec_properties: Dict[Text, Any]) -> None:
        
        print("Starting executor")
        print(input_dict)
        print(output_dict)
        print(exec_properties)
        print("Executor ran")


class MyCustomComponentSpec(ComponentSpec):
    PARAMETERS = {
        'input_config': ExecutionParameter(type=example_gen_pb2.Input),
        'output_config': ExecutionParameter(type=example_gen_pb2.Output)
    }
    INPUTS = {
        'input_params': ChannelParameter(type=standard_artifacts.ExternalArtifact)
    }
    OUTPUTS = {
        'output': ChannelParameter(type=standard_artifacts.String, optional=True)
    }


class MyCustomComponent(BaseComponent):
    SPEC_CLASS = MyCustomComponentSpec
    EXECUTOR_SPEC = ExecutorClassSpec(MyCustomExecutor)

    def __init__(
        self,
        input_params: Optional[Channel] = None
    ):
        input_config = utils.make_default_input_config()
        output_config = example_gen_pb2.Output()
        output = None

        spec = MyCustomComponentSpec(
            input_config=input_config,
            output_config=output_config,
            input_params=input_params,
            output=output
        )

        super(MyCustomComponent, self).__init__(spec=spec)

if __name__ == '__main__':
  absl.logging.set_verbosity(absl.logging.DEBUG)
  BeamDagRunner().run(
      _create_pipeline(
          pipeline_name=_pipeline_name,
          pipeline_root=_pipeline_root,
          data_root=_data_root,
          metadata_path=_metadata_path))

To reproduce the code, just run with python my_file_name.py

I'm using these versions:

tfx=1.8.0
tensorflow=2.8.0
apache-beam=2.50.0
python=3.9

I would like to set up my custom component with one string. Can anybody help me?