Scheduler/Webserver should never use Internal API, so calling any method decorated with internal_api_call should still execute them locally
data = {"jsonrpc": "2.0", "method": method_name, "params": params_json}
        internal_api_endpoint = InternalApiConfig.get_internal_api_endpoint()
        response = requests.post(url=internal_api_endpoint, data=json.dumps(data), headers=headers)
        if response.status_code != 200:
            raise AirflowException(
                f"Got {response.status_code}:{response.reason} when sending the internal api request."
            )
        return response.content
    @wraps(func)
    def wrapper(*args, **kwargs) -> RT:
        use_internal_api = InternalApiConfig.get_use_internal_api()
        if not use_internal_api:
            return func(*args, **kwargs)
        from airflow.serialization.serialized_objects import BaseSerialization  
        bound = inspect.signature(func).bind(*args, **kwargs)
        arguments_dict = dict(bound.arguments)
        if "session" in arguments_dict:
            del arguments_dict["session"]
        if "cls" in arguments_dict:  
            del arguments_dict["cls"]
        args_json = json.dumps(BaseSerialization.serialize(arguments_dict))
        method_name = f"{func.__module__}.{func.__qualname__}"
        result = make_jsonrpc_request(method_name, args_json)
        return BaseSerialization.deserialize(json.loads(result))
    return wrapper